diff --git a/doc/manual/configuration.rst b/doc/manual/configuration.rst index 4ffa31ee..3b5d91d0 100644 --- a/doc/manual/configuration.rst +++ b/doc/manual/configuration.rst @@ -2560,7 +2560,7 @@ The default is ``[download, upload]``. ``managed`` This archive is managed, meaning the files can be iterated and deleted. This is required for the archive command to work. Only supported with the - ``file`` and ``http`` backends so far. + ``file``, ``http`` and ``gitea`` backends so far. ``cache`` Use this archive to cache downloaded artifacts from other archives. If a binary artifact was successfully downloaded from another archive it will @@ -2602,6 +2602,14 @@ file Use a local directory as binary artifact repository. The directory directory. The optional ``fileMode`` and ``directoryMode`` keys take the desired access modes as numeric value to override the default umask derived modes. +gitea Uses a `Gitea generic package registry`_ as binary artifact + repository. Only the HEAD, GET, PUT and DELETE methods are used. + The base server URL is given in ``url``, the registry owner (user + or organization) in ``owner`` and the generic package name in + ``package``. As with the ``http`` backend the credentials are part + of the URL. The optional ``sslVerify`` boolean key controls whether + to verify the SSL certificate and ``retries`` sets the number of + retries on transient errors. http Uses a HTTP server as binary artifact repository. The server has to support the HEAD, PUT and GET methods. The base URL is given in the ``url`` key. The optional ``sslVerify`` boolean key controls @@ -2680,6 +2688,37 @@ the anonymous access to the container can be used like this:: The ``flags: [download]`` makes sure that Bob does not try to upload artifacts in case other backends are configured too. +The ``gitea`` backend stores the artifacts in a Gitea generic package registry. +The artifacts are put below ``{url}/api/packages/{owner}/generic/{package}/`` +with one package version per artifact:: + + archive: + - + backend: gitea + url: "https://user:passw%40rd@gitea.example.com" + owner: "bob-artifacts" + package: "myproject" + flags: [download, upload, managed] + +As for the ``http`` backend the credentials for the HTTP basic authentication +are taken from the URL and have to be percent encoded. Gitea accepts a personal +access token instead of the password. Uploading requires the ``write:package`` +scope, the ``managed`` flag additionally needs ``read:package``. + +.. warning:: + The password will be part of the Jenkins job configuration. Anybody who can + read the jobs ``config.xml`` will be able to retrieve the password! + +.. note:: + The registry cannot enumerate its content. The managed operations of + :ref:`manpage-archive` therefore use the package API of the server + (``{url}/api/v1/packages/...``) to list the artifacts. As with the ``http`` + backend only the artifact itself is deleted, the build-id and fingerprint + files are left behind. The package version is dropped by the server as soon + as its last file is gone. + +.. _Gitea generic package registry: https://docs.gitea.com/usage/packages/generic + .. _configuration-config-archive-prepend-append: archive{Prepend,Append} diff --git a/pym/bob/archive.py b/pym/bob/archive.py index 94a24e11..6b5b5ed5 100644 --- a/pym/bob/archive.py +++ b/pym/bob/archive.py @@ -24,7 +24,8 @@ from .tty import stepAction, stepMessage, \ SKIPPED, EXECUTED, WARNING, INFO, TRACE, ERROR, IMPORTANT from .utils import asHexStr, removePath, isWindows, getBashPath, tarfileOpen, binStat -from .webdav import WebDav, WebdavError, WebdavNotFoundError, WebdavAlreadyExistsError +from .webdav import WebDav, WebdavError, WebdavNotFoundError, WebdavAlreadyExistsError, \ + getNetLoc from tempfile import mkstemp, NamedTemporaryFile, TemporaryFile, gettempdir import asyncio import concurrent.futures @@ -32,6 +33,7 @@ import errno import gzip import io +import json import os import os.path import shutil @@ -861,6 +863,31 @@ def __exit__(self, exc_type, exc_value, traceback): return False +def retryWebdavRequest(request, retries): + """Run a WebDav request, retrying transient transport errors.""" + while True: + try: + return request() + except (WebdavError, OSError) as e: + if retries == 0: raise + retries -= 1 + + +def getWebdavAudit(webdav, path, retries, extract): + """Read the audit trail of a remote artifact. + + Only the beginning of the artifact is fetched. If the audit trail is not + part of it yet, more data is requested until it can be extracted. + """ + downloader = retryWebdavRequest(lambda: webdav.getPartialDownloader(path), retries) + while True: + try: + return extract(io.BytesIO(downloader.get())) + except (EOFError, tarfile.ReadError): + # partial downloader reached EOF or could not extract the audit from the tarfile, so we get more data + retryWebdavRequest(lambda: downloader.more(), retries) + + class HttpArchive(BaseArchive): def __init__(self, spec): super().__init__(spec) @@ -874,16 +901,10 @@ def getArchiveName(self): return name url = self.__url - return urllib.parse.urlunparse((url.scheme, url.netloc, url.path, '', '', '')) + return urllib.parse.urlunparse((url.scheme, getNetLoc(url), url.path, '', '', '')) def __retry(self, request): - retries = self._retries - while True: - try: - return request() - except (WebdavError, OSError) as e: - if retries == 0: raise - retries -= 1 + return retryWebdavRequest(request, self._retries) def _canManage(self): return True @@ -899,7 +920,7 @@ def _makeParentDirs(self, path): def _remoteName(self, buildId, suffix): url = self.__url - return urllib.parse.urlunparse((url.scheme, url.netloc, self._makePath(buildId, suffix), '', '', '')) + return urllib.parse.urlunparse((url.scheme, getNetLoc(url), self._makePath(buildId, suffix), '', '', '')) def _exists(self, path): return self.__retry(lambda: self._webdav.exists(path)) @@ -952,18 +973,11 @@ def _stat(self, filename): return struct.pack('=dL', parsedate_to_datetime(stats['mdate']).timestamp(), stats['len']) def _getAudit(self, filename): - downloader = self.__retry(lambda: self._webdav.getPartialDownloader("/".join([self.__url.path, filename]))) - while True: - try: - file = io.BytesIO(downloader.get()) - return self._extractAudit(fileobj=file) - except (EOFError, tarfile.ReadError): - # partial downloader reached EOF or could not extract the audit from the tarfile, so we get more data - self.__retry(lambda: downloader.more()) - pass + return getWebdavAudit(self._webdav, "/".join([self.__url.path, filename]), + self._retries, lambda f: self._extractAudit(fileobj=f)) def getArchiveUri(self): - return self.__url.netloc + self.__url.path + return getNetLoc(self.__url) + self.__url.path class HttpDownloader: @@ -1211,6 +1225,216 @@ def __upload(self): raise ArtifactError(str(e)) +class GiteaArchive(BaseArchive): + """Bob artifact backend for the Gitea 'generic' package registry. + + Artifacts are stored using the generic package API: + + {url}/api/packages/{owner}/generic/{package}/{version}/{filename} + + with ``version`` and ``filename`` both derived from the build-id. All + artifact types belonging to the same key (``.tgz``, ``.buildid``, + ``.fprnt``) share a single package version. + + The registry is a plain HTTP server as far as Bob is concerned. Only + HEAD, GET, PUT and DELETE are used, so the WebDav class does the actual + transport. As with the http backend the HTTP basic authentication + credentials are part of the URL. Gitea accepts a personal access token in + place of the password. + + The managed operations additionally use the package API of the server + (``{url}/api/v1/packages/...``) because the registry itself cannot + enumerate its content. + """ + + # The package API does not return more entries than that per request. + PAGE_SIZE = 50 + + def __init__(self, spec): + super().__init__(spec) + self.__url = urllib.parse.urlparse(spec["url"]) + self.__owner = spec["owner"] + self.__package = spec["package"] + self._webdav = WebDav(self.__url, spec.get("sslVerify", True)) + self._retries = spec.get("retries", 1) + # Caches of the managed operations. They are filled on demand and are + # only relevant for the lifetime of a "bob archive" invocation. + self.__versions = None + self.__files = {} + + def __basePath(self): + return "/".join([self.__url.path.rstrip("/"), "api", "packages", + self.__owner, "generic", self.__package]) + + def getArchiveName(self): + name = super().getArchiveName() + if name: + return name + return urllib.parse.urlunparse((self.__url.scheme, getNetLoc(self.__url), + self.__basePath(), '', '', '')) + + def _canManage(self): + return True + + def __packagePath(self, version, name): + return "/".join([self.__basePath(), version, name]) + + def _makePath(self, buildId, suffix): + # One package version per artifact. The build-id/fingerprint/tarball + # files of an artifact all live in that single version. + packageResultId = buildIdToName(buildId) + return self.__packagePath(packageResultId, packageResultId + suffix) + + def _remoteName(self, buildId, suffix): + return urllib.parse.urlunparse((self.__url.scheme, getNetLoc(self.__url), + self._makePath(buildId, suffix), '', '', '')) + + def __retry(self, request): + return retryWebdavRequest(request, self._retries) + + def _exists(self, path): + return self.__retry(lambda: self._webdav.exists(path)) + + def _openDownloadFile(self, buildId, suffix): + path = self._makePath(buildId, suffix) + return self.__retry(lambda: HttpDownloader(self, self._webdav.download(path))) + + def _openUploadFile(self, buildId, suffix, overwrite): + path = self._makePath(buildId, suffix) + if overwrite: + # The generic registry refuses to overwrite an existing file but + # the meta data files (build-id, fingerprint) must be replaced. + # Delete a possibly existing file first. This is inherently racy: + # a concurrent upload may still squeeze in between the delete and + # the PUT below and let the latter fail with a conflict. + self.__retry(lambda: self._webdav.deletePath(path)) + elif self._exists(path): + raise ArtifactExistsError() + return HttpUploader(self, path, overwrite) + + def _putUploadFile(self, path, tmp, overwrite): + return self.__retry(lambda: self._webdav.upload(path, tmp, overwrite)) + + # + # Managed operations. The registry itself cannot enumerate its content. + # That is done by the package API of the server instead, which is not part + # of the registry but knows about all package types. + # + + def __apiPath(self, *parts): + return "/".join([self.__url.path.rstrip("/"), "api", "v1", "packages", + self.__owner, "generic", self.__package, *parts]) + + def __apiGet(self, path, query=None): + def request(): + with self._webdav.download(path, query=query) as reply: + return reply.read() + + try: + return json.loads(self.__retry(request).decode("utf-8")) + except (UnicodeDecodeError, ValueError) as e: + raise WebdavError("Invalid reply of package API: " + str(e)) + + def __allVersions(self): + """All package versions, that is all artifacts of the archive.""" + if self.__versions is None: + versions = [] + page = 1 + while True: + try: + reply = self.__apiGet(self.__apiPath(), + "page={}&limit={}".format(page, self.PAGE_SIZE)) + except WebdavNotFoundError: + # The package is created with the first upload. Until then + # the archive is simply empty. + break + try: + versions.extend(i["version"] for i in reply) + except (KeyError, TypeError): + raise WebdavError("Unexpected reply of package API") + if len(reply) < self.PAGE_SIZE: break + page += 1 + self.__versions = versions + + return self.__versions + + def __versionFiles(self, version): + """The files of a single package version, indexed by their name. + + A version holds at most the tarball, the build-id and the fingerprint + of one artifact, so the reply always fits into a single page. + """ + files = self.__files.get(version) + if files is None: + try: + reply = self.__apiGet(self.__apiPath(version, "files")) + except WebdavNotFoundError: + reply = [] + try: + files = { i["name"] : i for i in reply } + except (KeyError, TypeError): + raise WebdavError("Unexpected reply of package API") + self.__files[version] = files + + return files + + @staticmethod + def __splitPath(filename): + """Translate an archive command path back to version and file name. + + The archive command works on the "//.tgz" layout of the + file and http backends. Our version is the file name without the + separators and without the suffix. + """ + name = filename.replace("\\", "/").replace("/", "") + return name.rpartition(".")[0], name + + def _listDir(self, path): + prefix = path.replace("\\", "/").strip("/") + if prefix == ".": prefix = "" + versions = self.__allVersions() + + # The first two levels are just the first four characters of the + # version, that is of the build-id. + if prefix == "": + return sorted({ v[0:2] for v in versions }) + elif len(prefix) == 2: + return sorted({ v[2:4] for v in versions if v.startswith(prefix) }) + + # Only on the last level the files matter. Asking for them is + # required: uploaded live-build-ids and fingerprints create versions + # that hold no artifact at all. + prefix = prefix.replace("/", "") + return sorted(name[4:] for v in versions if v.startswith(prefix) + for name in self.__versionFiles(v)) + + def _stat(self, filename): + version, name = self.__splitPath(filename) + info = self.__versionFiles(version).get(name) + if info is None: + raise ArtifactNotFoundError() + # Unlike the other backends we get a real content hash instead of a + # modification time. + stat = info.get("sha256") + if not stat: + raise ArtifactError("Missing stats for file " + filename) + return stat + + def _getAudit(self, filename): + version, name = self.__splitPath(filename) + return getWebdavAudit(self._webdav, self.__packagePath(version, name), + self._retries, lambda f: self._extractAudit(fileobj=f)) + + def _delete(self, filename): + version, name = self.__splitPath(filename) + self.__retry(lambda: self._webdav.deletePath(self.__packagePath(version, name))) + # The server drops the version together with its last file. + self.__files.pop(version, None) + + def getArchiveUri(self): + return getNetLoc(self.__url) + self.__url.path + + class MultiArchive: def __init__(self, archives): self.__archives = archives @@ -1283,6 +1507,8 @@ def getSingleArchiver(recipes, archiveSpec): return CustomArchive(archiveSpec, recipes.envWhiteList()) elif archiveBackend == "azure": return AzureArchive(archiveSpec) + elif archiveBackend == "gitea": + return GiteaArchive(archiveSpec) elif archiveBackend == "none": return DummyArchive() elif archiveBackend == "__jenkins": diff --git a/pym/bob/input.py b/pym/bob/input.py index 2830f377..2558271a 100644 --- a/pym/bob/input.py +++ b/pym/bob/input.py @@ -3171,7 +3171,7 @@ def validate(self, data): class ArchiveValidator: def __init__(self): - self.__validTypes = schema.Schema({'backend': schema.Or('none', 'file', 'http', 'shell', 'azure')}, + self.__validTypes = schema.Schema({'backend': schema.Or('none', 'file', 'http', 'shell', 'azure', 'gitea')}, ignore_extra_keys=True) baseArchive = { 'backend' : str, @@ -3199,12 +3199,21 @@ def __init__(self): schema.Optional('key') : str, schema.Optional('sasToken"') : str, }) + giteaArchive = baseArchive.copy() + giteaArchive.update({ + 'url' : HttpUrlValidator(), + 'owner' : str, + 'package' : str, + schema.Optional('sslVerify') : bool, + schema.Optional('retries') : PositiveValidator(), + }) self.__backends = { 'none' : schema.Schema(baseArchive), 'file' : schema.Schema(fileArchive), 'http' : schema.Schema(httpArchive), 'shell' : schema.Schema(shellArchive), 'azure' : schema.Schema(azureArchive), + 'gitea' : schema.Schema(giteaArchive), } def validate(self, data): diff --git a/pym/bob/webdav.py b/pym/bob/webdav.py index bf9d8b4d..a8162932 100644 --- a/pym/bob/webdav.py +++ b/pym/bob/webdav.py @@ -18,6 +18,23 @@ class WebdavNotFoundError(WebdavError): class WebdavAlreadyExistsError(WebdavError): pass + +def getNetLoc(url): + """Get the network location of a parsed URL without the credentials. + + The user name and password of the HTTP basic authentication are part of + the URL. They are sent in the Authorization header instead and must be + kept out of the request URL and of anything that is shown to the user. + """ + netloc = url.netloc + if url.username is not None: + # urlparse() delimits the user info at the *last* '@'. Cut at the same + # one, otherwise a password with an unencoded '@' yields a bogus host. + netloc = netloc.rsplit('@', 1)[1] + + return netloc + + class WebDav: class PartialDownloader: @@ -39,7 +56,14 @@ def more(self, length=512*1024): def __init__(self, url, sslVerify=True): self.__url = url self.__connection = None - self.__context = None if sslVerify else sslNoVerifyContext() + self.__sslVerify = sslVerify + + @property + def __context(self): + # Create the SSL context on demand. Holding it as an attribute would + # render the object unpicklable, but the archive backends are sent to + # the up-/download executor processes. + return None if self.__sslVerify else sslNoVerifyContext() def getPartialDownloader(self, path, length=512*1024): return self.PartialDownloader(self, path, length) @@ -54,14 +78,12 @@ def _getHeaders(self): userPass.encode("utf-8")).decode("ascii") return headers - def _getURL(self, path): + def _getURL(self, path, query=None): # remove username and password from URI - netloc = self.__url.netloc - if self.__url.username is not None: - netloc = self.__url.netloc.split('@')[1] - - return urlunsplit((self.__url.scheme, netloc, path, - self.__url.query, self.__url.fragment)) + if query is None: + query = self.__url.query + return urlunsplit((self.__url.scheme, getNetLoc(self.__url), path, + query, self.__url.fragment)) def exists(self, path): req = urllib.request.Request (self._getURL(path), @@ -79,12 +101,12 @@ def exists(self, path): return False - def download(self, path, offset=None, length=None): + def download(self, path, offset=None, length=None, query=None): headers = self._getHeaders() if offset is not None and length is not None: headers.update({'Range': 'bytes={}-{}'.format(offset, offset + length - 1)}) - req = urllib.request.Request (self._getURL(path), + req = urllib.request.Request (self._getURL(path, query), headers=headers, method="GET") try: return urllib.request.urlopen (req, context=self.__context) @@ -104,7 +126,7 @@ def upload(self, path, buf, overwrite): length = str(buf.tell()) buf.seek(0) headers = self._getHeaders() - headers.update({'Content-Length': length}) + headers.update({'Content-Length': length, 'Content-Type': 'application/octet-stream'}) if not overwrite: headers.update({'If-None-Match': '*'}) @@ -119,6 +141,11 @@ def upload(self, path, buf, overwrite): if e.status == 412: # precondition failed -> lost race with other upload raise WebdavAlreadyExistsError() + if e.status == 409: + # Some servers (e.g. the Gitea package registry) do not honour + # "If-None-Match" but refuse to overwrite an existing file with + # a conflict instead. + raise WebdavAlreadyExistsError() raise WebdavError("PUT {} {}".format(e.status, e.reason)) except (http.client.HTTPException, OSError) as e: raise WebdavError(str(e)) @@ -193,11 +220,12 @@ def listdir(self, path): return dir_infos def delete(self, filename): - base_path = self.__url.path # create a full path - filepath = '/'.join([base_path, filename.strip('/')]) + self.deletePath('/'.join([self.__url.path, filename.strip('/')])) + + def deletePath(self, path): headers = self._getHeaders() - req = urllib.request.Request (self._getURL(filepath), + req = urllib.request.Request (self._getURL(path), headers=headers, method="DELETE") status = reason = None try: diff --git a/test/unit/test_archive.py b/test/unit/test_archive.py index 06c9e8d7..c08daf98 100644 --- a/test/unit/test_archive.py +++ b/test/unit/test_archive.py @@ -4,13 +4,16 @@ # SPDX-License-Identifier: GPL-3.0-or-later from binascii import hexlify -from tempfile import NamedTemporaryFile, TemporaryDirectory +from tempfile import NamedTemporaryFile, TemporaryDirectory, TemporaryFile from unittest import TestCase, skipIf from unittest.mock import patch import asyncio import base64 import gzip +import hashlib import http.server +import json +import pickle import posixpath import os, os.path import socketserver @@ -22,10 +25,12 @@ import sys from mocks.http_server import HttpServerMock -from bob.archive import DummyArchive, HttpArchive, getArchiver -from bob.errors import BuildError +from bob.archive import DummyArchive, HttpArchive, GiteaArchive, getArchiver, \ + ArtifactExistsError +from bob.errors import BobError, BuildError from bob.utils import runInEventLoop, getProcessPoolExecutor -from bob.webdav import WebdavError +from bob.webdav import WebdavError, WebdavNotFoundError, WebdavAlreadyExistsError, \ + getNetLoc DOWNLOAD_ARITFACT = b'\x00'*20 NOT_EXISTS_ARTIFACT = b'\x01'*20 @@ -692,6 +697,51 @@ def testUnauthorized(self): run(archive.downloadPackage(DummyStep(), b'\x00'*20, "unused", "unused", executor=self.executor)) self.assertEqual(run(archive.downloadLocalLiveBuildId(DummyStep(), b'\x00'*20, executor=self.executor)), None) + def testNoCredentialsInMessages(self): + """The URL credentials must never show up in user visible strings. + + Everything that is derived from the URL ends up in log messages and + therefore in build logs and CI consoles. + """ + spec = { } + self._setArchiveSpec(spec) + archive = HttpArchive(spec) + + for name, uri in [ + ("getArchiveName", archive.getArchiveName()), + ("getArchiveUri", archive.getArchiveUri()), + ("_remoteName", archive._remoteName(DOWNLOAD_ARITFACT, ".tgz")), + ]: + with self.subTest(method=name): + self.assertNotIn(self.PASSWORD, uri) + self.assertNotIn(urllib.parse.quote(self.PASSWORD), uri) + self.assertNotIn("@", uri) + # ...but the host must still be there to be of any use + self.assertIn("{}:{}".format(self.ip, self.port), uri) + + +class TestGetNetLoc(TestCase): + """Unit tests for the URL credential sanitizer.""" + + def testNoCredentials(self): + url = urllib.parse.urlparse("https://host.test:8443/path") + self.assertEqual(getNetLoc(url), "host.test:8443") + + def testCredentialsRemoved(self): + url = urllib.parse.urlparse("https://user:pass@host.test:8443/path") + self.assertEqual(getNetLoc(url), "host.test:8443") + + def testPasswordWithAtSign(self): + """urlparse() delimits at the *last* '@'. We must cut at the same one.""" + url = urllib.parse.urlparse("https://user:p@ssw@rd@host.test/path") + self.assertEqual(url.hostname, "host.test") + self.assertEqual(getNetLoc(url), "host.test") + + def testIPv6(self): + url = urllib.parse.urlparse("https://user:pass@[::1]:8443/path") + self.assertEqual(getNetLoc(url), "[::1]:8443") + + @skipIf(sys.platform.startswith("win"), "requires POSIX platform") class TestCustomArchive(BaseTester, TestCase): @@ -791,3 +841,613 @@ def testRetriesAudit(self): archive = self._getHttpArchiveInstance(srv.port) with self.assertRaises(WebdavError): archive._getAudit(filepath) + + +def createGiteaHandler(repoPath, args, expectedAuth=None): + """Mock of the Gitea 'generic' package registry. + + Requests use the generic package layout + + /api/packages//generic/// + + which this handler translates to the on-disk layout used by the + BaseTester helpers (``///-1``) so + that all the shared upload/download assertions apply unchanged. + + The package API that the managed operations use is served too: + + /api/v1/packages//generic/[//files] + + If ``expectedAuth`` is given it is the full ``Authorization`` header value + (i.e. ``"Basic ..."``) that the client must send; anything else is answered + with ``401``. + """ + + class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, *a): + pass + + def _authOk(self): + if expectedAuth is None: + return True + if self.headers.get('Authorization') == expectedAuth: + return True + self.send_response(401, "Unauthorized") + self.end_headers() + return False + + def _diskPath(self): + fname = self.path.rsplit("/", 1)[-1] + for suffix in (".tgz", ".buildid", ".fprnt"): + if fname.endswith(suffix): + break + else: + return None + stem = fname[:-len(suffix)] # -1 + ident = stem[:-len("-1")] # (strip archive generation) + return os.path.join(repoPath, ident[0:2], ident[2:4], + stem[4:] + suffix) + + def _maybeFail(self): + if args.get("retries", 0) > 0: + args["retries"] -= 1 + self.send_error(500, "flaky") + return True + return False + + def _allVersions(self): + """All package versions, derived from the on-disk layout.""" + versions = set() + for root, dirs, files in os.walk(repoPath): + l2 = os.path.basename(root) + l1 = os.path.basename(os.path.dirname(root)) + if len(l1) != 2 or len(l2) != 2: continue + for f in files: + stem, _, ext = f.rpartition(".") + if ext in ("tgz", "buildid", "fprnt"): + versions.add(l1 + l2 + stem) + return sorted(versions) + + def _versionFiles(self, version): + ident = version[:-len("-1")] + files = [] + for suffix in (".tgz", ".buildid", ".fprnt"): + path = os.path.join(repoPath, ident[0:2], ident[2:4], + ident[4:] + "-1" + suffix) + if not os.path.isfile(path): continue + with open(path, "rb") as f: + data = f.read() + files.append({ "name" : version + suffix, "size" : len(data), + "sha256" : hashlib.sha256(data).hexdigest() }) + return files + + def _apiGet(self): + """Serve the package API. Returns False for ordinary requests.""" + path, _, query = self.path.partition("?") + if not path.startswith("/api/v1/packages/"): + return False + + if args.get("badApi"): + # a server that does not speak the package API at all + self.send_response(200) + self.send_header("Content-type", "text/html") + self.end_headers() + self.wfile.write(b"no idea what you want") + return True + + parts = path.strip("/").split("/") + versions = self._allVersions() + if len(parts) == 6: + if not versions: + # the package is created with the first upload + self.send_error(404, "package does not exist"); return True + query = urllib.parse.parse_qs(query) + limit = int(query.get("limit", ["50"])[0]) + first = (int(query.get("page", ["1"])[0]) - 1) * limit + body = [ { "version" : v } for v in versions[first:first+limit] ] + elif len(parts) == 8 and parts[7] == "files": + body = self._versionFiles(parts[6]) + else: + self.send_error(404, "not found"); return True + + data = json.dumps(body).encode("utf-8") + self.send_response(200) + self.send_header("Content-type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + return True + + def do_HEAD(self): + if not self._authOk(): return + if self._maybeFail(): return + path = self._diskPath() + if path is None or not os.path.exists(path): + self.send_error(404, "not found"); return + self.send_response(200) + self.send_header("Content-type", "application/octet-stream") + self.send_header("Content-Length", str(os.path.getsize(path))) + self.end_headers() + + def do_GET(self): + if not self._authOk(): return + if self._maybeFail(): return + if self._apiGet(): return + path = self._diskPath() + try: + with open(path, "rb") as f: + data = f.read() + except FileNotFoundError: + self.send_error(404, "not found"); return + except OSError: + self.send_error(500, "internal error"); return + self.send_response(200) + self.send_header("Content-type", "application/octet-stream") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_PUT(self): + length = int(self.headers.get('Content-Length', 0)) + content = self.rfile.read(length) + if not self._authOk(): return + if self._maybeFail(): return + path = self._diskPath() + # The generic registry refuses to overwrite an existing file. + if os.path.exists(path): + self.send_response(409); self.end_headers(); return + try: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb") as f: + f.write(content) + except OSError: + self.send_error(500, "internal error"); return + self.send_response(201); self.end_headers() + + def do_DELETE(self): + if not self._authOk(): return + if self._maybeFail(): return + path = self._diskPath() + if path and os.path.exists(path): + os.unlink(path) + self.send_response(204) + else: + self.send_response(404) + self.end_headers() + + return Handler + + +class TestGiteaArchive(BaseTester, TestCase): + + def setUp(self): + super().setUp() + self.args = {"retries": 0} + self.httpd = socketserver.ThreadingTCPServer(("localhost", 0), + createGiteaHandler(self.repo.name, self.args)) + self.ip, self.port = self.httpd.server_address + self.server = threading.Thread(target=self.httpd.serve_forever) + self.server.daemon = True + self.server.start() + + def tearDown(self): + self.httpd.shutdown() + self.httpd.server_close() + super().tearDown() + + def _setArchiveSpec(self, spec): + spec['name'] = "gitea" + spec['backend'] = "gitea" + spec["url"] = "http://{}:{}".format(self.ip, self.port) + spec["owner"] = "bob-artifacts" + spec["package"] = "test" + + def testRemoteName(self): + """The build-id maps to the generic package layout.""" + a = GiteaArchive({"backend":"gitea", "url":"https://gitea.example", + "owner":"o", "package":"p"}) + bid = bytes.fromhex("00112233445566778899aabbccddeeff00112233") + self.assertEqual(a._remoteName(bid, ".tgz"), + "https://gitea.example/api/packages/o/generic/p/" + "00112233445566778899aabbccddeeff00112233-1/" + "00112233445566778899aabbccddeeff00112233-1.tgz") + + def testArchiveName(self): + """Without an explicit name the package base URL identifies the archive.""" + spec = {"backend":"gitea", "url":"https://gitea.example/git", + "owner":"o", "package":"p"} + self.assertEqual(GiteaArchive(spec).getArchiveName(), + "https://gitea.example/git/api/packages/o/generic/p") + self.assertEqual(GiteaArchive(dict(spec, name="my-archive")).getArchiveName(), + "my-archive") + + +class TestArchivePickle(TestCase): + """The archive objects are passed to the up-/download executor processes + and must therefore be picklable. Verifying the SSL certificate or not must + not make a difference.""" + + def _checkPickle(self, spec): + for sslVerify in [True, False]: + archive = getArchiver(DummyRecipeSet(dict(spec, sslVerify=sslVerify))) + self.assertIsNotNone(pickle.loads(pickle.dumps(archive))) + + def testHttp(self): + self._checkPickle({"backend":"http", "url":"https://server.test/archive"}) + + def testGitea(self): + self._checkPickle({"backend":"gitea", "url":"https://gitea.example", + "owner":"o", "package":"p"}) + + +class TestGiteaAuthArchive(BaseTester, TestCase): + """Same as above but the mock server requires HTTP basic authentication. + The credentials are part of the URL. Gitea takes a personal access token in + place of the password.""" + + USER = "alice" + TOKEN = "s3cr3t-token" + + def setUp(self): + super().setUp() + self.args = {"retries": 0} + expectedAuth = "Basic " + base64.b64encode( + (self.USER + ":" + self.TOKEN).encode("utf-8")).decode("ascii") + self.httpd = socketserver.ThreadingTCPServer(("localhost", 0), + createGiteaHandler(self.repo.name, self.args, expectedAuth)) + self.ip, self.port = self.httpd.server_address + self.server = threading.Thread(target=self.httpd.serve_forever) + self.server.daemon = True + self.server.start() + + def tearDown(self): + self.httpd.shutdown() + self.httpd.server_close() + super().tearDown() + + def _url(self, user, password): + return "http://{}:{}@{}:{}".format(user, password, self.ip, self.port) + + def _setArchiveSpec(self, spec): + spec['backend'] = "gitea" + spec["url"] = self._url(self.USER, self.TOKEN) + spec["owner"] = "o" + spec["package"] = "p" + + def testWrongTokenFailsUpload(self): + """A wrong token must fail the upload before the artifact body is sent.""" + archive = GiteaArchive({"backend":"gitea", "owner":"o", "package":"p", + "url":self._url(self.USER, "wrong-token")}) + archive.wantUploadLocal(True) + with TemporaryDirectory() as tmp: + audit = os.path.join(tmp, "audit.json.gz") + content = os.path.join(tmp, "workspace") + with open(audit, "wb") as f: + f.write(b"AUDIT") + os.mkdir(content) + with open(os.path.join(content, "data"), "wb") as f: + f.write(b"DATA") + with self.assertRaises(BuildError) as cm: + run(archive.uploadPackage(DummyStep(), UPLOAD1_ARTIFACT, audit, + content, executor=self.executor)) + self.assertIn("401", str(cm.exception)) + + def testNoCredentialsInMessages(self): + """The URL credentials must never show up in user visible strings. + + getArchiveName() in particular feeds _namedErrorString() and is thus + printed on every plain "artifact not found" of a cache miss. + """ + spec = { } + self._setArchiveSpec(spec) + archive = GiteaArchive(spec) + + for method, uri in [ + ("getArchiveName", archive.getArchiveName()), + ("getArchiveUri", archive.getArchiveUri()), + ("_remoteName", archive._remoteName(DOWNLOAD_ARITFACT, ".tgz")), + ]: + with self.subTest(method=method): + self.assertNotIn(self.TOKEN, uri) + self.assertNotIn(self.USER, uri) + self.assertNotIn("@", uri) + # ...but the host must still be there to be of any use + self.assertIn("{}:{}".format(self.ip, self.port), uri) + + +class TestGiteaArchiveRetries(Base, TestCase): + + def setUp(self): + super().setUp() + self.VALID_FILE = self._createArtifact(VALID_ARTIFACT, valid_data=True) + + def _getArchive(self, port, retries): + spec = {'backend':'gitea', 'name':'gitea', 'owner':'o', 'package':'p', + 'retries':retries, 'url':"http://localhost:{}".format(port)} + return getArchiver(DummyRecipeSet(spec)) + + def _startServer(self, retries): + args = {"retries": retries} + httpd = socketserver.ThreadingTCPServer(("localhost", 0), + createGiteaHandler(self.repo.name, args)) + thread = threading.Thread(target=httpd.serve_forever) + thread.daemon = True + thread.start() + return httpd, httpd.server_address[1] + + def _download(self, archive): + archive.wantDownloadLocal(True) + with TemporaryDirectory() as tmp: + audit = os.path.join(tmp, "audit.json.gz") + content = os.path.join(tmp, "workspace") + return run(archive.downloadPackage(DummyStep(), VALID_ARTIFACT, + audit, content, executor=self.executor)) + + def _testRetries(self, r): + # server fails exactly 'r' times -> download succeeds within retries + httpd, port = self._startServer(r) + try: + self.assertTrue(self._download(self._getArchive(port, r))) + finally: + httpd.shutdown(); httpd.server_close() + + # server fails one more time than retries -> download fails (no throw) + httpd, port = self._startServer(r + 1) + try: + self.assertFalse(self._download(self._getArchive(port, r))) + finally: + httpd.shutdown(); httpd.server_close() + + def testRetriesWithNoRetries(self): + self._testRetries(0) + + def testRetriesWithOneRetry(self): + self._testRetries(1) + + def testRetriesWithMultipleRetries(self): + self._testRetries(3) + + +class TestGiteaManagedArchive(Base, TestCase): + """The managed operations that back the "bob archive" command. + + They are served by the package API of the server because the registry + itself cannot enumerate its content. + """ + + def setUp(self): + super().setUp() + self.args = {"retries": 0} + self.httpd = socketserver.ThreadingTCPServer(("localhost", 0), + createGiteaHandler(self.repo.name, self.args)) + self.addCleanup(self.httpd.server_close) + self.addCleanup(self.httpd.shutdown) + thread = threading.Thread(target=self.httpd.serve_forever) + thread.daemon = True + thread.start() + + def _archive(self): + spec = {'backend':'gitea', 'name':'gitea', 'owner':'o', 'package':'p', + 'flags':['download', 'upload', 'managed'], + 'url':"http://localhost:{}".format(self.httpd.server_address[1])} + return getArchiver(DummyRecipeSet(spec)) + + @staticmethod + def _scanPath(bid, suffix=".tgz"): + """The path of an artifact as the archive command sees it.""" + bid = hexlify(bid).decode("ascii") + return posixpath.join(bid[0:2], bid[2:4], bid[4:] + "-1" + suffix) + + def testCanManage(self): + self.assertTrue(self._archive().canManage()) + + def testEmptyArchive(self): + """Nothing was uploaded yet, so the package does not even exist.""" + self.assertEqual(self._archive().listDir("."), []) + + def testListDir(self): + self._createArtifact(DOWNLOAD_ARITFACT) + bid = hexlify(DOWNLOAD_ARITFACT).decode("ascii") + archive = self._archive() + + self.assertEqual(archive.listDir("."), [bid[0:2]]) + self.assertEqual(archive.listDir(bid[0:2]), [bid[2:4]]) + self.assertEqual(archive.listDir(posixpath.join(bid[0:2], bid[2:4])), + [bid[4:] + "-1.tgz"]) + + def testListDirWithoutArtifact(self): + """Live-build-ids create package versions that hold no artifact. + + Deriving the file names from the version names alone would make the + scanner stat a tarball that does not exist. + """ + self._createBuildId(UPLOAD1_ARTIFACT) + bid = hexlify(UPLOAD1_ARTIFACT).decode("ascii") + entries = self._archive().listDir(posixpath.join(bid[0:2], bid[2:4])) + self.assertEqual(entries, [bid[4:] + "-1.buildid"]) + + def testListDirPaginated(self): + """The package API returns at most 50 versions per request.""" + bids = [ bytes([i]) + b'\x42'*19 for i in range(60) ] + for bid in bids: + self._createBuildId(bid) + self.assertEqual(self._archive().listDir("."), + sorted(hexlify(bid).decode("ascii")[0:2] for bid in bids)) + + def _sha256(self, name): + with open(name, "rb") as f: + return hashlib.sha256(f.read()).hexdigest() + + def testStat(self): + """The file hash of the registry is the change indicator.""" + name = self._createArtifact(DOWNLOAD_ARITFACT) + path = self._scanPath(DOWNLOAD_ARITFACT) + old = self._archive().stat(path) + self.assertEqual(old, self._sha256(name)) + + # a changed artifact must be detected + with open(name, "ab") as f: + f.write(b'\x00') + new = self._archive().stat(path) + self.assertNotEqual(new, old) + self.assertEqual(new, self._sha256(name)) + + def testStatNotFound(self): + self._createArtifact(DOWNLOAD_ARITFACT) + with self.assertRaises(BobError): + self._archive().stat(self._scanPath(NOT_EXISTS_ARTIFACT)) + + def testGetAudit(self): + self._createArtifact(VALID_ARTIFACT, valid_data=True) + audit = self._archive().getAudit(self._scanPath(VALID_ARTIFACT)) + self.assertIsNotNone(audit) + self.assertEqual(audit.getArtifact().getMetaData(), "1") + + def testDelete(self): + name = self._createArtifact(DOWNLOAD_ARITFACT) + self._archive().deleteFile(self._scanPath(DOWNLOAD_ARITFACT)) + self.assertFalse(os.path.exists(name)) + + def testDeleteNotFound(self): + """Deleting a file that is already gone is not an error.""" + self._createArtifact(DOWNLOAD_ARITFACT) + self._archive().deleteFile(self._scanPath(NOT_EXISTS_ARTIFACT)) + + def testBrokenApiReply(self): + """A server that does not speak the package API must not throw up.""" + self._createArtifact(DOWNLOAD_ARITFACT) + self.args["badApi"] = True + with self.assertRaises(BobError): + self._archive().listDir(".") + + +def createGiteaStatusHandler(responses): + """Mock Gitea registry that answers each HTTP method with a scripted result. + + ``responses`` maps the HTTP method ("HEAD"/"GET"/"PUT"/"DELETE") to either + ``("status", code)`` to return that status code or ``("close",)`` to read + the request and then drop the connection without any response (simulating a + reverse-proxy or server that closes mid-upload). An unlisted method yields + ``404``. + """ + + class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, *a): + pass + + def _handle(self): + length = int(self.headers.get('Content-Length', 0) or 0) + if length: + self.rfile.read(length) + action = responses.get(self.command, ("status", 404)) + if action[0] == "close": + self.close_connection = True + self.wfile.close() + return + self.send_response(action[1]) + self.end_headers() + + do_HEAD = _handle + do_GET = _handle + do_PUT = _handle + do_DELETE = _handle + + return Handler + + +class TestGiteaArchiveErrors(Base, TestCase): + """Directly exercise the individual error-handling branches of the + GiteaArchive HTTP methods using a mock server with scripted responses.""" + + BUILD_ID = bytes.fromhex("00112233445566778899aabbccddeeff00112233") + + def _archive(self, responses): + self.responses = responses + self.httpd = socketserver.ThreadingTCPServer(("localhost", 0), + createGiteaStatusHandler(responses)) + self.addCleanup(self.httpd.server_close) + self.addCleanup(self.httpd.shutdown) + thread = threading.Thread(target=self.httpd.serve_forever) + thread.daemon = True + thread.start() + port = self.httpd.server_address[1] + return GiteaArchive({"backend": "gitea", "owner": "o", "package": "p", + "retries": 0, "url": "http://localhost:{}".format(port)}) + + def _upload(self, archive): + path = archive._makePath(self.BUILD_ID, ".tgz") + with TemporaryFile() as tmp: + tmp.write(b"DATA") + archive._putUploadFile(path, tmp, False) + + def testCanManageAndUri(self): + archive = self._archive({}) + self.assertTrue(archive._canManage()) + self.assertIn("localhost", archive.getArchiveUri()) + + def testDownloadNotFound(self): + archive = self._archive({"GET": ("status", 404)}) + with self.assertRaises(WebdavNotFoundError): + archive._openDownloadFile(self.BUILD_ID, ".tgz") + + def testDownloadHttpError(self): + archive = self._archive({"GET": ("status", 403)}) + with self.assertRaises(WebdavError): + archive._openDownloadFile(self.BUILD_ID, ".tgz") + + def testExistsByHead(self): + # the HEAD preflight finds the artifact -> no upload + archive = self._archive({"HEAD": ("status", 200)}) + with self.assertRaises(ArtifactExistsError): + archive._openUploadFile(self.BUILD_ID, ".tgz", False) + + def testExistsHttpError(self): + # HEAD with an unexpected status + archive = self._archive({"HEAD": ("status", 400)}) + with self.assertRaises(WebdavError): + archive._openUploadFile(self.BUILD_ID, ".tgz", False) + + def testUploadConflict(self): + # HEAD says "missing" but the PUT hits a 409 race. Gitea refuses to + # overwrite an existing file with a conflict. + archive = self._archive({"PUT": ("status", 409)}) + with self.assertRaises(WebdavAlreadyExistsError): + self._upload(archive) + + def testUploadPreconditionFailed(self): + archive = self._archive({"PUT": ("status", 412)}) + with self.assertRaises(WebdavAlreadyExistsError): + self._upload(archive) + + def testUploadHttpError(self): + archive = self._archive({"PUT": ("status", 400)}) + with self.assertRaises(WebdavError): + self._upload(archive) + + def testUploadUnexpectedStatus(self): + # a 2xx status that is not one of the accepted success codes + archive = self._archive({"PUT": ("status", 205)}) + with self.assertRaises(WebdavError): + self._upload(archive) + + def testUploadConnectionClosed(self): + # server drops the connection during upload + archive = self._archive({"PUT": ("close",)}) + with self.assertRaises(WebdavError): + self._upload(archive) + + def testOverwriteDeletesFirst(self): + # the file is removed before the upload because Gitea would refuse to + # overwrite it + archive = self._archive({"DELETE": ("status", 204)}) + self.assertIsNotNone(archive._openUploadFile(self.BUILD_ID, ".tgz", True)) + + def testOverwriteDeleteMissing(self): + # nothing to delete is not an error + archive = self._archive({"DELETE": ("status", 404)}) + self.assertIsNotNone(archive._openUploadFile(self.BUILD_ID, ".tgz", True)) + + def testOverwriteDeleteHttpError(self): + archive = self._archive({"DELETE": ("status", 400)}) + with self.assertRaises(WebdavError): + archive._openUploadFile(self.BUILD_ID, ".tgz", True)