269 lines
9.8 KiB
Python
269 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from trusted_provider_receipts import (
|
|
PROVIDER_RECEIPT_SCHEMA,
|
|
PROVIDER_RECEIPT_VERSION,
|
|
ProviderAuthenticationError,
|
|
TrustedProviderPolicy,
|
|
TrustedProviderReceiptResolver,
|
|
build_hmac_authentication,
|
|
canonical_json_bytes,
|
|
canonical_sha256,
|
|
)
|
|
|
|
SOURCE = "a" * 40
|
|
TREE = "b" * 40
|
|
NOW = datetime(2026, 9, 9, 5, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _receipt(result: dict, **overrides):
|
|
payload = {
|
|
"schema": PROVIDER_RECEIPT_SCHEMA,
|
|
"version": PROVIDER_RECEIPT_VERSION,
|
|
"provider_identity": "aios.tests-runner",
|
|
"provider_operation": "tests.run",
|
|
"operation_run_id": "run-123",
|
|
"result_reference": "runs/run-123/junit.xml",
|
|
"content_sha256": canonical_sha256(result),
|
|
"source_oid": SOURCE,
|
|
"tree_oid": TREE,
|
|
"environment": "linux-isolated",
|
|
"issued_at": "2026-09-09T04:59:00+00:00",
|
|
"result": result,
|
|
"authentication": {
|
|
"kind": "root-owned-store-v1",
|
|
"store_id": "tests",
|
|
},
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def _write_receipt(root: Path, ref: str, payload: dict, *, mode=0o600):
|
|
path = root / ref
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
# All provider-store directories are authenticated and non-writable.
|
|
cur = root
|
|
os.chmod(cur, 0o700)
|
|
for part in Path(ref).parts[:-1]:
|
|
cur = cur / part
|
|
os.chmod(cur, 0o700)
|
|
path.write_bytes(canonical_json_bytes(payload))
|
|
os.chmod(path, mode)
|
|
return path
|
|
|
|
|
|
def _policy(root: Path, **overrides):
|
|
values = dict(
|
|
evidence_kind="tests",
|
|
provider_identity="aios.tests-runner",
|
|
store_id="tests",
|
|
store_root=root,
|
|
allowed_operations=frozenset({"tests.run"}),
|
|
expected_owner_uid=os.getuid(),
|
|
allowed_directory_modes=frozenset({0o700}),
|
|
allowed_receipt_modes=frozenset({0o600}),
|
|
max_age_seconds=300,
|
|
)
|
|
values.update(overrides)
|
|
return TrustedProviderPolicy(**values)
|
|
|
|
|
|
def _resolver(root: Path, **policy_overrides):
|
|
return TrustedProviderReceiptResolver(
|
|
{"tests": _policy(root, **policy_overrides)},
|
|
now_fn=lambda: NOW,
|
|
)
|
|
|
|
|
|
def test_valid_root_owned_receipt_is_authenticated(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
result = {"passed": True, "summary": {"passed": 10, "failed": 0}}
|
|
_write_receipt(root, "run-123/receipt.json", _receipt(result))
|
|
|
|
resolved = _resolver(root).resolve_evidence(
|
|
"tests",
|
|
{"receipt_ref": "run-123/receipt.json"},
|
|
source_oid=SOURCE,
|
|
tree_oid=TREE,
|
|
)
|
|
assert resolved.result == result
|
|
assert resolved.provider_receipt["provider_identity"] == "aios.tests-runner"
|
|
assert resolved.provider_receipt["authentication"]["kind"] == "root-owned-store-v1"
|
|
assert len(resolved.provider_receipt["receipt_file_sha256"]) == 64
|
|
|
|
|
|
def test_caller_supplied_receipt_body_is_denied(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
with pytest.raises(ProviderAuthenticationError, match="CALLER_SUPPLIED_PROVIDER_RECEIPT_DENIED"):
|
|
_resolver(root).resolve_evidence(
|
|
"tests",
|
|
{"receipt": _receipt({"passed": True})},
|
|
source_oid=SOURCE,
|
|
tree_oid=TREE,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("ref", ["/abs/receipt.json", "../escape.json", "a/../../escape.json"])
|
|
def test_absolute_or_traversal_receipt_ref_is_denied(tmp_path: Path, ref: str):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_REF_"):
|
|
_resolver(root).resolve("tests", ref, source_oid=SOURCE, tree_oid=TREE)
|
|
|
|
|
|
def test_symlink_receipt_is_denied(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
target = root / "real.json"
|
|
target.write_text("{}", encoding="utf-8")
|
|
os.chmod(target, 0o600)
|
|
link = root / "receipt.json"
|
|
link.symlink_to(target.name)
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_SYMLINK_DENIED"):
|
|
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|
|
|
|
|
|
def test_world_writable_receipt_is_denied(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
path = _write_receipt(root, "receipt.json", _receipt({"passed": True}), mode=0o666)
|
|
assert path.exists()
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_MODE_INVALID|PROVIDER_RECEIPT_WRITABLE_DENIED"):
|
|
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|
|
|
|
|
|
def test_world_writable_store_is_denied(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
_write_receipt(root, "receipt.json", _receipt({"passed": True}))
|
|
os.chmod(root, 0o777)
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_STORE_MODE_INVALID|PROVIDER_STORE_WRITABLE_DENIED"):
|
|
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|
|
|
|
|
|
def test_wrong_owner_is_denied(monkeypatch, tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
path = _write_receipt(root, "receipt.json", _receipt({"passed": True}))
|
|
real_lstat = os.lstat
|
|
|
|
def fake_lstat(p):
|
|
st = real_lstat(p)
|
|
if Path(p) == path:
|
|
values = list(st)
|
|
values[4] = st.st_uid + 1
|
|
return os.stat_result(values)
|
|
return st
|
|
|
|
monkeypatch.setattr(os, "lstat", fake_lstat)
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_OWNER_MISMATCH"):
|
|
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|
|
|
|
|
|
def test_provider_identity_mismatch_is_denied(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
_write_receipt(root, "receipt.json", _receipt({"passed": True}, provider_identity="evil"))
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_IDENTITY_MISMATCH"):
|
|
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|
|
|
|
|
|
def test_operation_not_allowlisted_is_denied(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
_write_receipt(root, "receipt.json", _receipt({"passed": True}, provider_operation="shell.exec"))
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_OPERATION_DENIED"):
|
|
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|
|
|
|
|
|
def test_content_digest_mismatch_is_denied(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
result = {"passed": True}
|
|
_write_receipt(root, "receipt.json", _receipt(result, content_sha256="0" * 64))
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_CONTENT_SHA256_MISMATCH"):
|
|
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "error"),
|
|
[("source_oid", "PROVIDER_SOURCE_OID_MISMATCH"), ("tree_oid", "PROVIDER_TREE_OID_MISMATCH")],
|
|
)
|
|
def test_source_or_tree_mismatch_is_denied(tmp_path: Path, field: str, error: str):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
_write_receipt(root, "receipt.json", _receipt({"passed": True}, **{field: "c" * 40}))
|
|
with pytest.raises(ProviderAuthenticationError, match=error):
|
|
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|
|
|
|
|
|
def test_stale_receipt_is_denied(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
_write_receipt(root, "receipt.json", _receipt({"passed": True}, issued_at="2026-09-09T04:00:00+00:00"))
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_EXPIRED"):
|
|
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|
|
|
|
|
|
def test_request_body_cannot_override_authenticated_result(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
result = {"passed": True, "count": 10}
|
|
_write_receipt(root, "receipt.json", _receipt(result))
|
|
with pytest.raises(ProviderAuthenticationError, match="EVIDENCE_REQUEST_CONTENT_MISMATCH"):
|
|
_resolver(root).resolve_evidence(
|
|
"tests",
|
|
{"receipt_ref": "receipt.json", "passed": False, "count": 10},
|
|
source_oid=SOURCE,
|
|
tree_oid=TREE,
|
|
)
|
|
|
|
|
|
def test_hmac_authenticated_receipt_passes_and_bad_mac_fails(tmp_path: Path):
|
|
root = tmp_path / "store"
|
|
root.mkdir(mode=0o700)
|
|
key = b"test-only-secret"
|
|
key_id = "provider-tests-hmac"
|
|
result = {"passed": True, "summary": "ok"}
|
|
receipt = _receipt(result)
|
|
receipt["authentication"] = {"kind": "hmac-sha256-v1", "key_id": key_id}
|
|
receipt["authentication"] = build_hmac_authentication(
|
|
receipt,
|
|
key_id=key_id,
|
|
key=key,
|
|
)
|
|
_write_receipt(root, "receipt.json", receipt)
|
|
|
|
policy = _policy(
|
|
root,
|
|
authentication_kind="hmac-sha256-v1",
|
|
hmac_key_id=key_id,
|
|
)
|
|
resolver = TrustedProviderReceiptResolver(
|
|
{"tests": policy},
|
|
hmac_key_resolver=lambda requested: key if requested == key_id else b"",
|
|
now_fn=lambda: NOW,
|
|
)
|
|
resolved = resolver.resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|
|
assert resolved.result["passed"] is True
|
|
|
|
bad = json.loads((root / "receipt.json").read_text(encoding="utf-8"))
|
|
bad["authentication"]["mac_hex"] = "0" * 64
|
|
(root / "receipt.json").write_bytes(canonical_json_bytes(bad))
|
|
os.chmod(root / "receipt.json", 0o600)
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_HMAC_MISMATCH"):
|
|
resolver.resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
|