[upload] Add HTTP(S) and Canonical upload targets

Adds two new upload targets - a generic http(s) target and a port of the
existing ubuntu policy upload to a `canonical` upload target.

Note that this also changes the `upload-method` option to
`--upload-http-method` and also removes `auto` as a valid choice,
leaving just `put` and `post`, as it is assumed that the `auto` logic
would now be redundant for independently implemented vendor targets that
subclass the generic http target.

Signed-off-by: Jake Hunsaker <[email protected]>

sos/policies/distros/ubuntu.py

@@ -6,7 +6,6 @@
 #
 # See the LICENSE file in the source distribution for further information.
 
-import os
 
 from sos.report.plugins import UbuntuPlugin
 from sos.policies.distros.debian import DebianPolicy
@@ -26,10 +25,6 @@ class UbuntuPolicy(DebianPolicy):
     os_release_file = ''
     PATH = "/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games" \
            + ":/usr/local/sbin:/usr/local/bin:/snap/bin"
-    _upload_url = "https://files.support.canonical.com/uploads/"
-    _upload_user = "ubuntu"
-    _upload_password = "ubuntu"
-    _upload_method = 'put'
 
     def __init__(self, sysroot=None, init=None, probe_runtime=True,
                  remote_exec=None):
@@ -66,22 +61,4 @@ def dist_version(self):
         except (IOError, ValueError):
             return False
 
-    def get_upload_https_auth(self, user=None, password=None):
-        if self.upload_url.startswith(self._upload_url):
-            return (self._upload_user, self._upload_password)
-        return super().get_upload_https_auth()
-
-    def get_upload_url_string(self):
-        if self.upload_url.startswith(self._upload_url):
-            return "Canonical Support File Server"
-        return self._get_obfuscated_upload_url(self.get_upload_url())
-
-    def get_upload_url(self):
-        if not self.upload_url or self.upload_url.startswith(self._upload_url):
-            if not self.upload_archive_name:
-                return self._upload_url
-            fname = os.path.basename(self.upload_archive_name)
-            return self._upload_url + fname
-        return super().get_upload_url()
-
 # vim: set et ts=4 sw=4 :

sos/upload/__init__.py

@@ -45,7 +45,7 @@ class SoSUpload(SoSComponent):
         'upload_directory': None,
         'upload_user': None,
         'upload_pass': None,
-        'upload_method': 'auto',
+        'upload_http_method': 'put',
         'upload_no_ssl_verify': False,
         'upload_target': None,
         'upload_s3_endpoint': None,
@@ -154,8 +154,8 @@ def add_parser_options(cls, parser):
                                 help="Username to authenticate with")
         upload_grp.add_argument("--upload-pass", default=None,
                                 help="Password to authenticate with")
-        upload_grp.add_argument("--upload-http-method", default='auto',
-                                choices=['auto', 'put', 'post'],
+        upload_grp.add_argument("--upload-http-method", default='put',
+                                choices=['put', 'post'],
                                 help="HTTP method to use for uploading")
         upload_grp.add_argument("--upload-no-ssl-verify", default=False,
                                 action='store_true',

sos/upload/targets/canonical.py

@@ -0,0 +1,31 @@
+# Copyright 2025, Jake Hunsaker <[email protected]>
+
+# This file is part of the sos project: https://github.com/sosreport/sos
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# version 2 of the GNU General Public License.
+#
+# See the LICENSE file in the source distribution for further information.
+
+from sos.upload.targets import SoSUploadTarget
+from sos.upload.targets.http import HttpUpload
+
+
+class CanonicalUpload(SoSUploadTarget):
+    """
+    The upload target for providing archives to the Canonical support team.
+    """
+    target_id = 'canonical'
+    target_name = 'Canonical Support File Server'
+    prompt_for_credentials = False
+
+    def upload(self):
+        _uploader = HttpUpload(
+            self.opts,
+            url='https://files.support.canonical.com/uploads/',
+            username='ubuntu',
+            password='ubuntu',
+            method='put'
+        )
+        return _uploader.upload()

sos/upload/targets/http.py

@@ -0,0 +1,95 @@
+# Copyright 2025, Jake Hunsaker <[email protected]>
+
+# This file is part of the sos project: https://github.com/sosreport/sos
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# version 2 of the GNU General Public License.
+#
+# See the LICENSE file in the source distribution for further information.
+
+import requests
+
+from sos.upload.targets import SoSUploadTarget
+from sos.utilities import TIMEOUT_DEFAULT
+
+
+class HttpUpload(SoSUploadTarget):
+    """
+    Generic handler for HTTP(S) uploads. This target can be used standalone
+    for adhoc uploads from user-provided addresses, and also be used as the
+    backend for a vendor-provided target, so that individual vendors do not
+    need to (but still may) rewrite http upload mechanisms in their own target.
+    """
+    target_id = 'http'
+    target_protocols = ['http', 'https']
+    prompt_for_credentials = True
+
+    def __init__(self, options, url=None, username=None, password=None,
+                 method=None, no_ssl_verify=None):
+        if username:
+            options.upload_user = username
+        if password:
+            options.upload_pass = password
+        if url:
+            options.upload_url = url
+        if no_ssl_verify:
+            options.upload_no_ssl_verify = no_ssl_verify
+        self.method = method or options.upload_http_method
+        self.http_headers = {}
+        super().__init__(options)
+
+    def set_http_auth_headers(self, headers):
+        if not isinstance(headers, dict):
+            raise TypeError(
+                f"Headers for http(s) upload must be provided as a dict, got "
+                f"{headers.__class__}"
+            )
+        self.http_headers = headers
+
+    def set_http_method(self, method):
+        """Set the http method to use from the requests library"""
+        if method not in ['put', 'post']:
+            raise ValueError(
+                f"HTTP(S) upload method must be 'put' or 'post', not "
+                f"'{method}'"
+            )
+        self.method = method
+
+    def get_http_auth(self):
+        return requests.auth.HTTPBasicAuth(self.opts.upload_user,
+                                           self.opts.upload_pass)
+
+    def upload(self):
+        if self.method == 'put':
+            ret = self._upload_with_put()
+        else:
+            ret = self._upload_with_post()
+        if ret.status_code not in (200, 201):
+            if ret.status_code == 401:
+                raise Exception(
+                    "Authentication failed: invalid user credentials"
+                )
+            raise Exception(
+                f"Upload request returned {ret.status_code}: {ret.reason}"
+            )
+        return True
+
+    def _upload_with_put(self):
+        """Attempt an upload to an http(s) endpoint via a PUT"""
+        return requests.put(
+            self.opts.upload_url, data=self.upload_file,
+            auth=self.get_http_auth(),
+            verify=self.opts.upload_no_ssl_verify,
+            timeout=TIMEOUT_DEFAULT
+        )
+
+    def _upload_with_post(self):
+        files = {
+            'file': (self.upload_file.split('/')[-1], self.upload_file,
+                     self.http_headers)
+        }
+        return requests.post(
+            self.opts.upload_url, files=files, auth=self.get_http_auth(),
+            verify=self.opts.upload_no_ssl_verify, timeout=TIMEOUT_DEFAULT
+        )