Credentials support for ZIP and TAR

README.rst

@@ -145,6 +145,43 @@ The set of repositories to operate on can optionally be restricted by the type:
 If the command should work on multiple repositories make sure to pass only generic arguments which work for all of these repository types.
 
 
+Authorization for Fetching ZIP or TAR Repositories
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Credentials for ZIP and TAR repositories can be provided using environment variables.
+
+1. Define a Credentials Key in the Repository Configuration
+
+Add a ``credentials_key`` field in your repository configuration:
+
+.. code-block:: yaml
+
+   repositories:
+     contents/example:
+       type: zip
+       url: https://127.0.0.1/content.zip
+       credentials_key: <KEY>
+
+2. Set Environment Variables with Credentials
+
+Use appropriate environment variables to pass authentication details:
+
+- ``VCS_<KEY>_AUTHENTICATION_METHOD`` – Authentication method:  
+  - ``Basic`` for username and password.  
+  - ``Bearer`` or ``Token`` for token-based authentication.
+- ``VCS_<KEY>_USERNAME`` – Username for ``Basic`` authentication.
+- ``VCS_<KEY>_PASSWORD`` – Password for ``Basic`` authentication.
+- ``VCS_<KEY>_TOKEN`` – Token for ``Bearer`` or ``Token`` authentication.
+
+3. (Optional) Disable SSL Certificate Verification
+
+To disable SSL certificate verification, set the following environment variable:
+
+.. code-block:: bash
+
+   export VCS_IGNORE_SSL_CERTIFICATE=True
+
+
 How to install vcstool?
 =======================
 

vcstool/clients/tar.py

@@ -44,7 +44,7 @@ def import_(self, command):
 
         # download tarball
         try:
-            data = load_url(command.url, retry=command.retry)
+            data = load_url(command.url, retry=command.retry, credentials_key=command.credentials_key)
         except URLError as e:
             return {
                 'cmd': '',

vcstool/clients/vcs_base.py

@@ -2,6 +2,8 @@
 import socket
 import subprocess
 import time
+import ssl
+import base64
 from urllib.error import HTTPError
 from urllib.error import URLError
 from urllib.request import Request
@@ -13,6 +15,11 @@ class VcsClientBase(object):
     type = None
 
     def __init__(self, path):
+        ignore_certs = os.environ.get("VCS_IGNORE_SSL_CERTIFICATE", False)
+
+        if ignore_certs:
+            ssl._create_default_https_context = ssl._create_unverified_context
+
         self.path = path
 
     def __getattribute__(self, name):
@@ -78,15 +85,43 @@ def run_command(cmd, cwd, env=None):
     return result
 
 
-def load_url(url, retry=2, retry_period=1, timeout=10):
+def _add_credentials_to_request(request, credentials_key):
+    authentication_method = os.environ.get(f"VCS_{credentials_key}_AUTHENTICATION_METHOD", "Basic")
+    token = os.environ.get(f"VCS_{credentials_key}_TOKEN", "")
+    username = os.environ.get(f"VCS_{credentials_key}_USERNAME","")
+    password = os.environ.get(f"VCS_{credentials_key}_PASSWORD", "")
+
+    base64string = ""
+    if authentication_method == "Basic" and (username or password):
+        base64string = base64.b64encode(bytes(f"{username}:{password}", 'ascii')).decode('utf-8')
+    elif authentication_method == "Token" or authentication_method == "Bearer":
+        base64string = token
+
+    request.add_header("Authorization", f"{authentication_method} {base64string}")
+
+
+def load_url(url, retry=2, retry_period=1, timeout=10, credentials_key=None):
+    request = Request(url)
     try:
-        fh = urlopen(url, timeout=timeout)
+        if credentials_key:
+            _add_credentials_to_request(request, credentials_key)
+        fh = urlopen(request, timeout=timeout)
     except HTTPError as e:
         if e.code == 503 and retry:
             time.sleep(retry_period)
             return load_url(
                 url, retry=retry - 1, retry_period=retry_period,
                 timeout=timeout)
+        elif e.code == 401:
+            if not credentials_key:
+                e.msg += f". Credentials not provided. Add 'credentials_key' field in vcs file and " \
+                    f"VCS_<KEY>_AUTHENTICATION_METHOD, VCS_<KEY>_USERNAME, VCS_<KEY>_PASSWORD, VCS_<KEY>_TOKEN " \
+                    "environment variables to set up credentials."
+            else:
+                e.msg += f". Credentials invalid or missing. Set up authentication method with " \
+                    f"VCS_{credentials_key}_AUTHENTICATION_METHOD ('Basic', 'Bearer' or 'Token') and " \
+                    f"credentials with VCS_{credentials_key}_TOKEN or VCS_{credentials_key}_USERNAME " \
+                    f"and VCS_{credentials_key}_PASSWORD environment variables."
         e.msg += ' (%s)' % url
         raise
     except URLError as e:

vcstool/clients/zip.py

@@ -44,7 +44,7 @@ def import_(self, command):
 
         # download zipfile
         try:
-            data = load_url(command.url, retry=command.retry)
+            data = load_url(command.url, retry=command.retry, credentials_key=command.credentials_key)
         except URLError as e:
             return {
                 'cmd': '',

vcstool/commands/import_.py

@@ -24,7 +24,7 @@ class ImportCommand(Command):
     help = 'Import the list of repositories'
 
     def __init__(
-        self, args, url, version=None, recursive=False, shallow=False
+        self, args, url, version=None, recursive=False, shallow=False, credentials_key=None
     ):
         super(ImportCommand, self).__init__(args)
         self.url = url
@@ -34,6 +34,7 @@ def __init__(
         self.skip_existing = args.skip_existing
         self.recursive = recursive
         self.shallow = shallow
+        self.credentials_key = credentials_key
 
 
 def get_parser():
@@ -107,6 +108,8 @@ def get_repos_in_vcstool_format(repositories):
             repo['url'] = attributes['url']
             if 'version' in attributes:
                 repo['version'] = attributes['version']
+            if 'credentials_key' in attributes:
+                repo['credentials_key'] = attributes['credentials_key']
         except KeyError as e:
             print(
                 ansi('yellowf') + (
@@ -138,6 +141,8 @@ def get_repos_in_rosinstall_format(root):
             repo['url'] = attributes['uri']
             if 'version' in attributes:
                 repo['version'] = attributes['version']
+            if 'credentials_key' in attributes:
+                repo['credentials_key'] = attributes['credentials_key']
         except KeyError as e:
             print(
                 ansi('yellowf') + (
@@ -171,7 +176,8 @@ def generate_jobs(repos, args):
         command = ImportCommand(
             args, repo['url'],
             str(repo['version']) if 'version' in repo else None,
-            recursive=args.recursive, shallow=args.shallow)
+            recursive=args.recursive, shallow=args.shallow, 
+            credentials_key=repo['credentials_key'] if 'credentials_key' in repo else None)
         job = {'client': client, 'command': command}
         jobs.append(job)
     return jobs