| import contextlib |
| import hashlib |
| import io |
| import json |
| import os |
| import subprocess |
| import sys |
| import tempfile |
| import unittest |
| from pathlib import Path |
| from unittest.mock import patch |
|
|
| SIGNING_DIRECTORY = Path(__file__).resolve().parent |
| sys.path.insert(0, str(SIGNING_DIRECTORY)) |
| import notarize_with_akv as notary |
|
|
| CONFIGURATION = notary.NotarizationConfiguration( |
| "01234567-89ab-cdef-0123-456789abcdef", |
| "APPLEKEY01", |
| "notary-vault", |
| "example-signing-key", |
| "0123456789abcdef" * 2, |
| ) |
| ENVIRONMENT = { |
| "AZURE_KEYVAULT_NAME": CONFIGURATION.vault_name, |
| "APPLE_NOTARIZATION_AKV_KEY_NAME": CONFIGURATION.vault_key_name, |
| } |
| CREDENTIALS = { |
| "awsAccessKeyId": "temporary-access-key", |
| "awsSecretAccessKey": "temporary-secret-key", |
| "awsSessionToken": "temporary-session-token", |
| "bucket": "apple-notary-bucket", |
| "object": "uploads/release artifact.dmg", |
| } |
|
|
|
|
| def response(payload): |
| return io.BytesIO( |
| json.dumps(payload).encode() if isinstance(payload, dict) else payload |
| ) |
|
|
|
|
| def azure_response(payload): |
| return subprocess.CompletedProcess([], 0, stdout=json.dumps(payload), stderr="") |
|
|
|
|
| class NotarizationTest(unittest.TestCase): |
| def test_validates_the_apple_key_tags_and_pins_its_version(self) -> None: |
| metadata = { |
| "id": CONFIGURATION.versioned_key_id, |
| "apple_key_id": "APPLEKEY01", |
| "apple_issuer_id": CONFIGURATION.issuer_id, |
| } |
| with ( |
| patch.dict(os.environ, ENVIRONMENT, clear=True), |
| patch.object( |
| notary.subprocess, |
| "run", |
| return_value=azure_response(metadata), |
| ) as show, |
| ): |
| self.assertEqual( |
| notary.NotarizationConfiguration.from_environment(), |
| CONFIGURATION, |
| ) |
| show.return_value = azure_response({**metadata, "apple_key_id": None}) |
| with self.assertRaisesRegex(notary.NotarizationError, "apple-key-id tag"): |
| notary.NotarizationConfiguration.from_environment() |
| for issuer_id in (None, "", " "): |
| with self.subTest(issuer_id=issuer_id): |
| show.return_value = azure_response( |
| {**metadata, "apple_issuer_id": issuer_id} |
| ) |
| with self.assertRaisesRegex( |
| notary.NotarizationError, "apple-issuer-id tag" |
| ): |
| notary.NotarizationConfiguration.from_environment() |
| show.return_value = azure_response( |
| {**metadata, "apple_issuer_id": "invalid-issuer"} |
| ) |
| with self.assertRaisesRegex(notary.NotarizationError, "valid UUID"): |
| notary.NotarizationConfiguration.from_environment() |
| self.assertIn( |
| '{id:key.kid,apple_key_id:tags."apple-key-id",' |
| 'apple_issuer_id:tags."apple-issuer-id"}', |
| show.call_args.args[0], |
| ) |
|
|
| invalid_environment = { |
| **ENVIRONMENT, |
| "APPLE_NOTARIZATION_AKV_KEY_NAME": "../bad", |
| } |
| with ( |
| patch.dict(os.environ, invalid_environment, clear=True), |
| self.assertRaisesRegex(notary.NotarizationError, "key name"), |
| ): |
| notary.NotarizationConfiguration.from_environment() |
|
|
| def test_signs_the_notary_scoped_digest_with_the_pinned_key(self) -> None: |
| signature = bytes(range(64)) |
| payload = { |
| "kid": CONFIGURATION.versioned_key_id, |
| "value": notary.base64url_encode(signature), |
| } |
| with patch.object( |
| notary.subprocess, "run", return_value=azure_response(payload) |
| ) as sign: |
| token = notary.create_apple_jwt(CONFIGURATION, issued_at=1_780_000_000) |
| header, claims, encoded_signature = token.split(".") |
| self.assertEqual( |
| json.loads(notary.base64url_decode(header))["kid"], "APPLEKEY01" |
| ) |
| self.assertEqual( |
| json.loads(notary.base64url_decode(claims))["iss"], CONFIGURATION.issuer_id |
| ) |
| self.assertEqual( |
| json.loads(notary.base64url_decode(claims))["scope"], |
| ["/notary/v2"], |
| ) |
| self.assertEqual(notary.base64url_decode(encoded_signature), signature) |
| command = sign.call_args.args[0] |
| body = json.loads(command[command.index("--body") + 1]) |
| self.assertEqual(body["alg"], "ES256") |
| self.assertEqual( |
| notary.base64url_decode(body["value"]), |
| hashlib.sha256(f"{header}.{claims}".encode()).digest(), |
| ) |
|
|
| def test_rejects_wrong_key_versions_and_invalid_signatures(self) -> None: |
| cases = ( |
| ( |
| CONFIGURATION.versioned_key_id + "wrong", |
| bytes(64), |
| "unexpected key version", |
| ), |
| (CONFIGURATION.versioned_key_id, b"short", "64 JOSE"), |
| ) |
| for key_id, signature, message in cases: |
| payload = {"kid": key_id, "value": notary.base64url_encode(signature)} |
| with ( |
| self.subTest(message=message), |
| patch.object( |
| notary.subprocess, |
| "run", |
| return_value=azure_response(payload), |
| ), |
| self.assertRaisesRegex(notary.NotarizationError, message), |
| ): |
| notary.create_apple_jwt(CONFIGURATION, issued_at=1) |
|
|
| def test_uploads_polls_and_retains_sanitized_apple_diagnostics(self) -> None: |
| signature = azure_response( |
| { |
| "kid": CONFIGURATION.versioned_key_id, |
| "value": notary.base64url_encode(bytes(64)), |
| } |
| ) |
| responses = [ |
| response({"data": {"id": "submission-1", "attributes": CREDENTIALS}}), |
| response(b""), |
| response({"data": {"attributes": {"status": "In Progress"}}}), |
| response({"data": {"attributes": {"status": "Accepted"}}}), |
| response( |
| { |
| "data": { |
| "attributes": { |
| "developerLogUrl": "https://logs.example.com/log" |
| } |
| } |
| } |
| ), |
| response(b'{"status":"Accepted"}'), |
| ] |
| with tempfile.TemporaryDirectory() as directory: |
| artifact, log = ( |
| Path(directory) / "codex.zip", |
| Path(directory) / "notary-log.json", |
| ) |
| artifact.write_bytes(b"signed release binary") |
| output = io.StringIO() |
| with ( |
| patch.object(notary.subprocess, "run", return_value=signature), |
| patch.object( |
| notary.urllib.request, "urlopen", side_effect=responses |
| ) as requests, |
| patch.object(notary.time, "sleep") as sleep, |
| contextlib.redirect_stdout(output), |
| ): |
| submission = notary.notarize( |
| artifact, CONFIGURATION, max_wait_seconds=600, report_log=log |
| ) |
| self.assertEqual(submission, "submission-1") |
| self.assertEqual(json.loads(log.read_text()), {"status": "Accepted"}) |
| sleep.assert_called_once_with(10) |
| upload = requests.call_args_list[1].args[0] |
| self.assertEqual(upload.get_method(), "PUT") |
| self.assertNotIn("temporary-secret-key", upload.get_header("Authorization")) |
| self.assertNotIn("temporary-secret-key", output.getvalue()) |
|
|
|
|
| class NotarizationWrapperTest(unittest.TestCase): |
| def test_binary_and_dmg_wrappers_preserve_notarization_contracts( |
| self, |
| ) -> None: |
| with tempfile.TemporaryDirectory() as directory: |
| root, tools = Path(directory), Path(directory) / "tools" |
| tools.mkdir() |
| call_log = root / "calls.txt" |
| environment = { |
| **os.environ, |
| **ENVIRONMENT, |
| "PATH": f"{tools}:{os.environ['PATH']}", |
| "RUNNER_TEMP": str(root), |
| "CALL_LOG": str(call_log), |
| } |
| for name in ("az", "python3", "rcodesign"): |
| body = ( |
| "exit 0" |
| if name == "az" |
| else f'printf "{name} %s\\n" "$*" >> "$CALL_LOG"' |
| ) |
| executable = tools / name |
| executable.write_text( |
| f"#!/usr/bin/env bash\nset -euo pipefail\n{body}\n" |
| ) |
| executable.chmod(0o755) |
| for kind in ("binary", "dmg"): |
| with self.subTest(kind=kind): |
| artifact = root / ("codex.dmg" if kind == "dmg" else "codex") |
| artifact.write_bytes(b"signed release artifact") |
| subprocess.run( |
| [ |
| str( |
| SIGNING_DIRECTORY / f"notarize_macos_{kind}_with_akv.sh" |
| ), |
| f"--{kind}", |
| str(artifact), |
| "--report-dir", |
| str(root / "report"), |
| ], |
| env=environment, |
| check=True, |
| capture_output=True, |
| text=True, |
| ) |
| calls = call_log.read_text().splitlines() |
| self.assertIn("notarize_with_akv.py --file", calls[0]) |
| if kind == "dmg": |
| self.assertEqual(calls[1], f"rcodesign staple {artifact}") |
| else: |
| self.assertEqual(len(calls), 1) |
| self.assertEqual(list(root.rglob("*.p8")), []) |
| call_log.unlink() |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
