diff --git a/README.md b/README.md index efe5e7d..b14ab99 100644 --- a/README.md +++ b/README.md @@ -135,6 +135,13 @@ If you are using docker registry with a self signed ssl certificate, you can dis registry.py -l user:pass -r https://example.com:5000 --no-validate-ssl ``` +## Nexus docker registry + +Add --digest-method flag + +``` +registry.py -l user:pass -r https://example.com:5000 --digest-method GET +``` ## Important notes: @@ -175,7 +182,7 @@ You are very welcome to contribute to this script. Of course, when making change please include your changes into `test.py` and run tests to check that your changes do not break existing functionality. -For tests to work, install `mock` library +For tests to work, more libraries are needed ``` pip install -r requirements-ci.txt ``` diff --git a/registry.py b/registry.py index 72ee77f..1d2b443 100755 --- a/registry.py +++ b/registry.py @@ -170,6 +170,7 @@ class Registry: self.no_validate_ssl = False self.http = None self.last_error = None + self.digest_method = "HEAD" def parse_login(self, login): if login is not None: @@ -188,7 +189,7 @@ class Registry: @staticmethod - def create(host, login, no_validate_ssl): + def create(host, login, no_validate_ssl, digest_method = "HEAD"): r = Registry() (r.username, r.password) = r.parse_login(login) @@ -199,6 +200,7 @@ class Registry: r.hostname = host r.no_validate_ssl = no_validate_ssl r.http = Requests() + r.digest_method = digest_method return r @@ -262,10 +264,10 @@ class Registry: def get_tag_digest(self, image_name, tag): image_headers = self.send("/v2/{0}/manifests/{1}".format( - image_name, tag), method="HEAD") + image_name, tag), method=self.digest_method) if image_headers is None: - print(" tag digest not found: {0}".format(self.last_error)) + print(" tag digest not found: {0}. Try --digest-method=GET".format(self.last_error)) return None tag_digest = image_headers.headers['Docker-Content-Digest'] @@ -491,6 +493,13 @@ for more detail on garbage collection read here: nargs='?', metavar='Hours') + parser.add_argument( + '--digest-method', + help=('Use HEAD for standard docker registry or GET for NEXUS'), + default='HEAD', + metavar="HEAD|GET" + ) + return parser.parse_args(args) @@ -620,7 +629,8 @@ def main_loop(args): args.login = username + ':' + password - registry = Registry.create(args.host, args.login, args.no_validate_ssl) + registry = Registry.create(args.host, args.login, args.no_validate_ssl, + args.digest_method) registry.auth_schemes = get_auth_schemes(registry,'/v2/_catalog') diff --git a/test.py b/test.py index 3b0e696..0fb3f3f 100644 --- a/test.py +++ b/test.py @@ -264,6 +264,35 @@ class TestListDigest(unittest.TestCase): response, 'sha256:85295b0e7456a8fbbc886722b483f87f2bff553fa0beeaf37f5d807aff7c1e52') self.assertEqual(self.registry.last_error, None) + def test_get_digest_nexus_ok(self): + self.registry.http.reset_return_value(status_code=200, + text=('{' + '"schemaVersion": 2,\n ' + '"mediaType": "application/vnd.docker.distribution.manifest.v2+json"' + '"digest": "sha256:357ea8c3d80bc25792e010facfc98aee5972ebc47e290eb0d5aea3671a901cab"' + )) + + self.registry.http.return_value.headers = { + 'Content-Length': '4935', + 'Docker-Content-Digest': 'sha256:85295b0e7456a8fbbc886722b483f87f2bff553fa0beeaf37f5d807aff7c1e52', + 'X-Content-Type-Options': 'nosniff' + } + + self.registry.digest_method = "GET" + response = self.registry.get_tag_digest('image1', '0.1.300') + self.registry.http.request.assert_called_with( + "GET", + "http://testdomain.com/v2/image1/manifests/0.1.300", + auth=(None, None), + headers=self.registry.HEADERS, + verify=True + ) + + self.assertEqual( + response, 'sha256:85295b0e7456a8fbbc886722b483f87f2bff553fa0beeaf37f5d807aff7c1e52') + self.assertEqual(self.registry.last_error, None) + + def test_invalid_status_code(self): self.registry.http.reset_return_value(400) response = self.registry.get_tag_digest('image1', '0.1.300') @@ -660,7 +689,8 @@ class TestArgParser(unittest.TestCase): "--no-validate-ssl", "--delete-all", "--layers", - "--delete-by-hours", "24"] + "--delete-by-hours", "24", + "--digest-method", "GET"] args = parse_args(args_list) self.assertTrue(args.delete) self.assertTrue(args.layers) @@ -674,6 +704,13 @@ class TestArgParser(unittest.TestCase): self.assertEqual(args.host, "hostname") self.assertEqual(args.keep_tags, ["keep1", "keep2"]) self.assertEqual(args.delete_by_hours, "24") + self.assertEqual(args.digest_method, "GET") + + def test_default_args(self): + args_list = ["-r", "hostname", + "-l", "loginstring"] + args = parse_args(args_list) + self.assertEqual(args.digest_method, "HEAD") if __name__ == '__main__':