130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from ops import attest_gcp_reasoning_compute as attestation
|
|
|
|
|
|
def reasoning_receipt(path: Path) -> Path:
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema": attestation.REASONING_SCHEMA,
|
|
"status": "pass",
|
|
"generated_at_utc": "2026-07-15T01:00:00+00:00",
|
|
"completed_at_utc": "2026-07-15T01:01:00+00:00",
|
|
"target_database": "teleo_clone_current_test",
|
|
"source_compute": "teleo-prod-1",
|
|
"errors": [],
|
|
"posted_to_telegram": False,
|
|
"database_write_attempted": False,
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
def compute_identity() -> dict:
|
|
return {
|
|
"project": "teleo-501523",
|
|
"project_number": "785938879453",
|
|
"instance": "teleo-prod-1",
|
|
"instance_id": "123456789",
|
|
"zone": "europe-west6-a",
|
|
"zone_resource": "projects/123456789/zones/europe-west6-a",
|
|
"network_resource": "projects/123456789/networks/teleo-staging-net",
|
|
"private_ip": "10.61.0.2",
|
|
"service_account": attestation.DEFAULT_RESTORE_SERVICE_ACCOUNT,
|
|
"expected_private_network": "projects/teleo-501523/global/networks/teleo-staging-net",
|
|
"checks": {
|
|
"project": True,
|
|
"project_number": True,
|
|
"expected_network_project": True,
|
|
"instance": True,
|
|
"instance_id": True,
|
|
"zone": True,
|
|
"network": True,
|
|
"private_ip": True,
|
|
"service_account": True,
|
|
},
|
|
}
|
|
|
|
|
|
def test_attestation_binds_reasoning_hash_to_live_metadata(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
receipt_path = reasoning_receipt(tmp_path / "reasoning.json")
|
|
observed = {}
|
|
|
|
def fake_identity(project, instance, zone, network, service_account, *, timeout):
|
|
observed.update(
|
|
{
|
|
"project": project,
|
|
"instance": instance,
|
|
"zone": zone,
|
|
"network": network,
|
|
"service_account": service_account,
|
|
"timeout": timeout,
|
|
}
|
|
)
|
|
return compute_identity()
|
|
|
|
monkeypatch.setattr(attestation, "gce_compute_identity", fake_identity)
|
|
|
|
result = attestation.attest_reasoning_compute(
|
|
reasoning_receipt_path=receipt_path,
|
|
restore_run_id="gcp-restore-current-test123",
|
|
target_database="teleo_clone_current_test",
|
|
generated_at_utc="2026-07-15T01:01:05+00:00",
|
|
)
|
|
|
|
assert result["status"] == "pass"
|
|
assert result["checks"]["gce_metadata_identity"] is True
|
|
assert result["reasoning_receipt"]["sha256"] == attestation.sha256_file(receipt_path)
|
|
assert result["compute"]["instance_id"] == "123456789"
|
|
assert observed == {
|
|
"project": "teleo-501523",
|
|
"instance": "teleo-prod-1",
|
|
"zone": "europe-west6-a",
|
|
"network": "projects/teleo-501523/global/networks/teleo-staging-net",
|
|
"service_account": attestation.DEFAULT_RESTORE_SERVICE_ACCOUNT,
|
|
"timeout": 10.0,
|
|
}
|
|
|
|
|
|
def test_attestation_rejects_claimed_wrong_compute_before_metadata(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
receipt_path = reasoning_receipt(tmp_path / "reasoning.json")
|
|
payload = json.loads(receipt_path.read_text())
|
|
payload["source_compute"] = "untrusted-vm"
|
|
receipt_path.write_text(json.dumps(payload) + "\n", encoding="utf-8")
|
|
monkeypatch.setattr(
|
|
attestation,
|
|
"gce_compute_identity",
|
|
lambda *args, **kwargs: pytest.fail("metadata must not be queried"),
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="source_compute"):
|
|
attestation.attest_reasoning_compute(
|
|
reasoning_receipt_path=receipt_path,
|
|
restore_run_id="gcp-restore-current-test123",
|
|
target_database="teleo_clone_current_test",
|
|
)
|
|
|
|
|
|
def test_attestation_rejects_send_or_write_receipt(tmp_path: Path) -> None:
|
|
receipt_path = reasoning_receipt(tmp_path / "reasoning.json")
|
|
payload = json.loads(receipt_path.read_text())
|
|
payload["posted_to_telegram"] = True
|
|
payload["database_write_attempted"] = True
|
|
receipt_path.write_text(json.dumps(payload) + "\n", encoding="utf-8")
|
|
|
|
with pytest.raises(ValueError, match=r"no_database_write.*no_telegram_send"):
|
|
attestation.attest_reasoning_compute(
|
|
reasoning_receipt_path=receipt_path,
|
|
restore_run_id="gcp-restore-current-test123",
|
|
target_database="teleo_clone_current_test",
|
|
)
|