diff --git a/scripts/sync_schemas.py b/scripts/sync_schemas.py index 5629ceaa..1fa9de25 100755 --- a/scripts/sync_schemas.py +++ b/scripts/sync_schemas.py @@ -16,6 +16,12 @@ The target version comes from `src/adcp/ADCP_VERSION`. If that version's bundle is not published, sync falls back to `latest.tgz` (the dev snapshot). +Environment variables: + ADCP_BASE_URL Override the protocol host (default: https://adcontextprotocol.org). + Set to point at a fixture CDN for cross-SDK CI or pre-release testing. + Trailing slashes are stripped automatically. Do NOT include "/protocol". + ADCP_SKIP_SIGNATURE Set to "1" to skip Sigstore verification and trust the SHA-256 only. + Usage: python scripts/sync_schemas.py # sync schemas + skills python scripts/sync_schemas.py --no-skills # schemas only (e.g. drift checks) @@ -41,7 +47,17 @@ CACHE_DIR = REPO_ROOT / "schemas" / "cache" SKILLS_DIR = REPO_ROOT / "skills" VERSION_FILE = REPO_ROOT / "src" / "adcp" / "ADCP_VERSION" -BUNDLE_BASE_URL = "https://adcontextprotocol.org/protocol" +_ADCP_BASE = os.environ.get("ADCP_BASE_URL", "https://adcontextprotocol.org").rstrip("/") +# Reject overrides ending in /protocol — appending our own /protocol below +# would silently produce //protocol and 404 against any sensible CDN. Fail +# loud at module import so the typo surfaces immediately. +if _ADCP_BASE.endswith("/protocol"): + raise ValueError( + f"ADCP_BASE_URL={_ADCP_BASE!r} ends with '/protocol'. The script " + "appends '/protocol' itself — pass only the protocol host " + "(e.g. https://adcontextprotocol.org)." + ) +BUNDLE_BASE_URL = _ADCP_BASE + "/protocol" USER_AGENT = "adcp-python-sdk/3.0" # Sigstore keyless verification identity. Must match the upstream release @@ -118,9 +134,7 @@ def fetch_signature_sidecars(version: str) -> tuple[bytes | None, bytes | None]: return sig, crt -def verify_cosign_signature( - tgz_bytes: bytes, sig_bytes: bytes, crt_bytes: bytes -) -> None: +def verify_cosign_signature(tgz_bytes: bytes, sig_bytes: bytes, crt_bytes: bytes) -> None: """Verify the bundle with `cosign verify-blob`. Raises RuntimeError if cosign is not installed or verification fails. @@ -193,9 +207,7 @@ def replace_cache_from_bundle(bundle_root: Path) -> int: """ schemas_src = bundle_root / "schemas" if not schemas_src.is_dir(): - raise RuntimeError( - f"Bundle missing expected directory: {bundle_root.name}/schemas/" - ) + raise RuntimeError(f"Bundle missing expected directory: {bundle_root.name}/schemas/") if CACHE_DIR.exists(): shutil.rmtree(CACHE_DIR) @@ -274,6 +286,8 @@ def main() -> None: args = parser.parse_args() target_version = get_target_adcp_version() + if "ADCP_BASE_URL" in os.environ: + print(f" ! ADCP_BASE_URL override active: {_ADCP_BASE}") print(f"Syncing AdCP protocol bundle from {BUNDLE_BASE_URL}...") print(f"Target version: {target_version}") print(f"Schema cache: {CACHE_DIR}") @@ -283,9 +297,7 @@ def main() -> None: try: print(f"Fetching {target_version}.tgz + checksum...") - tgz_bytes, expected_sha, effective_version = fetch_bundle_with_fallback( - target_version - ) + tgz_bytes, expected_sha, effective_version = fetch_bundle_with_fallback(target_version) except (HTTPError, URLError) as exc: print(f"\n✗ Failed to download bundle: {exc}", file=sys.stderr) sys.exit(1) @@ -311,10 +323,7 @@ def main() -> None: sys.exit(1) if sig_bytes is None or crt_bytes is None: - print( - f" ! No Sigstore sidecars for adcp-{effective_version} " - "(checksum-only trust)" - ) + print(f" ! No Sigstore sidecars for adcp-{effective_version} " "(checksum-only trust)") else: try: verify_cosign_signature(tgz_bytes, sig_bytes, crt_bytes) diff --git a/tests/test_sync_schemas.py b/tests/test_sync_schemas.py index 9a098b35..f9834011 100644 --- a/tests/test_sync_schemas.py +++ b/tests/test_sync_schemas.py @@ -124,9 +124,7 @@ def test_no_manifest_returns_zero(self, capsys: pytest.CaptureFixture[str]) -> N assert result == 0 assert "No manifest.json" in capsys.readouterr().out - def test_empty_skills_list_returns_zero( - self, capsys: pytest.CaptureFixture[str] - ) -> None: + def test_empty_skills_list_returns_zero(self, capsys: pytest.CaptureFixture[str]) -> None: with tempfile.TemporaryDirectory() as tmp_str: tmp = Path(tmp_str) bundle_root = _make_bundle(tmp, manifest_skills=[]) @@ -179,15 +177,11 @@ def test_previous_snapshot_created_on_update(self) -> None: existing.mkdir() (existing / "SKILL.md").write_text("# Old Brand") - bundle_root = _make_bundle( - tmp, skills={"adcp-brand": {"SKILL.md": "# New Brand"}} - ) + bundle_root = _make_bundle(tmp, skills={"adcp-brand": {"SKILL.md": "# New Brand"}}) sync_skills_from_bundle(bundle_root, skills_dir) assert (skills_dir / "adcp-brand" / "SKILL.md").read_text() == "# New Brand" - assert (skills_dir / "adcp-brand.previous" / "SKILL.md").read_text() == ( - "# Old Brand" - ) + assert (skills_dir / "adcp-brand.previous" / "SKILL.md").read_text() == ("# Old Brand") def test_previous_snapshot_replaced_on_second_update(self) -> None: with tempfile.TemporaryDirectory() as tmp_str: @@ -204,14 +198,10 @@ def test_previous_snapshot_replaced_on_second_update(self) -> None: existing.mkdir() (existing / "SKILL.md").write_text("# Old Brand") - bundle_root = _make_bundle( - tmp, skills={"adcp-brand": {"SKILL.md": "# New Brand"}} - ) + bundle_root = _make_bundle(tmp, skills={"adcp-brand": {"SKILL.md": "# New Brand"}}) sync_skills_from_bundle(bundle_root, skills_dir) - assert (skills_dir / "adcp-brand.previous" / "SKILL.md").read_text() == ( - "# Old Brand" - ) + assert (skills_dir / "adcp-brand.previous" / "SKILL.md").read_text() == ("# Old Brand") def test_local_only_skill_untouched(self) -> None: with tempfile.TemporaryDirectory() as tmp_str: @@ -263,9 +253,7 @@ def test_path_traversal_slash_in_name_rejected(self) -> None: with pytest.raises(RuntimeError, match="Unsafe skill name rejected"): sync_skills_from_bundle(bundle_root, skills_dir) - def test_non_string_name_skipped( - self, capsys: pytest.CaptureFixture[str] - ) -> None: + def test_non_string_name_skipped(self, capsys: pytest.CaptureFixture[str]) -> None: with tempfile.TemporaryDirectory() as tmp_str: tmp = Path(tmp_str) bundle_root = _make_bundle( @@ -280,9 +268,7 @@ def test_non_string_name_skipped( assert count == 1 # only the valid string entry is synced assert "Skipping non-string" in capsys.readouterr().out - def test_missing_bundle_skill_dir_skipped( - self, capsys: pytest.CaptureFixture[str] - ) -> None: + def test_missing_bundle_skill_dir_skipped(self, capsys: pytest.CaptureFixture[str]) -> None: with tempfile.TemporaryDirectory() as tmp_str: tmp = Path(tmp_str) # Manifest lists a skill that has no corresponding directory in the bundle @@ -322,9 +308,7 @@ def test_missing_bundle_skill_preserves_existing_dst( sync_skills_from_bundle(bundle_root, skills_dir) # dst must not be touched when src is absent - assert (skills_dir / "adcp-brand" / "SKILL.md").read_text() == ( - "# Existing Brand" - ) + assert (skills_dir / "adcp-brand" / "SKILL.md").read_text() == ("# Existing Brand") assert "missing in bundle" in capsys.readouterr().out def test_multiple_skills_synced(self) -> None: @@ -346,3 +330,62 @@ def test_multiple_skills_synced(self) -> None: assert (skills_dir / "adcp-brand" / "SKILL.md").exists() assert (skills_dir / "adcp-creative" / "SKILL.md").exists() assert (skills_dir / "call-adcp-agent" / "SKILL.md").exists() + + +# --------------------------------------------------------------------------- +# BUNDLE_BASE_URL env override (ADCP_BASE_URL) +# --------------------------------------------------------------------------- + + +class TestBundleBaseUrl: + def test_default_value(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Fresh load with env var absent — guards against shell having ADCP_BASE_URL set. + monkeypatch.delenv("ADCP_BASE_URL", raising=False) + fresh_spec = importlib.util.spec_from_file_location("sync_schemas_default", _SCRIPT) + assert fresh_spec is not None and fresh_spec.loader is not None + fresh_mod = importlib.util.module_from_spec(fresh_spec) + fresh_spec.loader.exec_module(fresh_mod) # type: ignore[union-attr] + assert fresh_mod.BUNDLE_BASE_URL == "https://adcontextprotocol.org/protocol" + + def test_env_override(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Fresh module load with ADCP_BASE_URL set to verify override is applied. + monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com") + fresh_spec = importlib.util.spec_from_file_location("sync_schemas_fresh", _SCRIPT) + assert fresh_spec is not None and fresh_spec.loader is not None + fresh_mod = importlib.util.module_from_spec(fresh_spec) + fresh_spec.loader.exec_module(fresh_mod) # type: ignore[union-attr] + assert fresh_mod.BUNDLE_BASE_URL == "https://fixture.example.com/protocol" + + def test_env_override_strips_trailing_slash(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Trailing slash on ADCP_BASE_URL must not produce "//protocol". + monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com/") + fresh_spec = importlib.util.spec_from_file_location("sync_schemas_fresh2", _SCRIPT) + assert fresh_spec is not None and fresh_spec.loader is not None + fresh_mod = importlib.util.module_from_spec(fresh_spec) + fresh_spec.loader.exec_module(fresh_mod) # type: ignore[union-attr] + assert fresh_mod.BUNDLE_BASE_URL == "https://fixture.example.com/protocol" + assert "//protocol" not in fresh_mod.BUNDLE_BASE_URL + + def test_env_override_rejects_protocol_suffix(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Override ending in /protocol would double-append. Fail loud at + # import rather than silently 404-ing later. + monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com/protocol") + fresh_spec = importlib.util.spec_from_file_location("sync_schemas_protocol_suffix", _SCRIPT) + assert fresh_spec is not None and fresh_spec.loader is not None + fresh_mod = importlib.util.module_from_spec(fresh_spec) + with pytest.raises(ValueError, match="ends with '/protocol'"): + fresh_spec.loader.exec_module(fresh_mod) # type: ignore[union-attr] + + def test_env_override_rejects_protocol_suffix_with_trailing_slash( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # Same guard, but with a trailing slash on the override — rstrip + # runs first, so the /protocol still trips the check. + monkeypatch.setenv("ADCP_BASE_URL", "https://fixture.example.com/protocol/") + fresh_spec = importlib.util.spec_from_file_location( + "sync_schemas_protocol_trailing", _SCRIPT + ) + assert fresh_spec is not None and fresh_spec.loader is not None + fresh_mod = importlib.util.module_from_spec(fresh_spec) + with pytest.raises(ValueError, match="ends with '/protocol'"): + fresh_spec.loader.exec_module(fresh_mod) # type: ignore[union-attr]