feat(security): add GitHub private backup evidence gate

This commit is contained in:
ogt
2026-06-26 17:45:00 +08:00
parent 59485d519c
commit 5d6b128854
10 changed files with 1621 additions and 0 deletions

View File

@@ -316,6 +316,9 @@ from src.services.docker_build_surface_inventory import (
from src.services.gitea_workflow_runner_health import (
load_latest_gitea_workflow_runner_health,
)
from src.services.github_target_private_backup_evidence_gate import (
load_latest_github_target_private_backup_evidence_gate,
)
from src.services.host_runaway_aiops_loop_readiness import (
load_latest_host_runaway_aiops_loop_readiness,
)
@@ -3109,6 +3112,33 @@ async def get_gitea_workflow_runner_health() -> dict[str, Any]:
) from exc
@router.get(
"/github-target-private-backup-evidence-gate",
response_model=dict[str, Any],
summary="取得 GitHub 私有備援 evidence gate",
description=(
"讀取最新已提交的 GitHub target private backup evidence gate"
"此端點不呼叫 GitHub / Gitea API、不建立 repo、不修改 visibility、"
"不同步 refs、不觸發 workflow、不切 GitHub primary、不讀取或保存 secret value。"
),
)
async def get_github_target_private_backup_evidence_gate() -> dict[str, Any]:
"""Return the latest read-only GitHub private backup evidence gate."""
try:
return await asyncio.to_thread(load_latest_github_target_private_backup_evidence_gate)
except FileNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(exc),
) from exc
except (json.JSONDecodeError, ValueError) as exc:
logger.error("github_target_private_backup_evidence_gate_invalid", error=str(exc))
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="GitHub 私有備援 evidence gate 快照無效",
) from exc
@router.get(
"/observability-contract-matrix",
response_model=dict[str, Any],

View File

@@ -0,0 +1,192 @@
"""GitHub target private backup evidence gate snapshot loader.
Loads the committed, read-only GitHub private backup evidence gate. This module
never calls GitHub / Gitea, creates repos, changes visibility, syncs refs,
triggers workflows, switches primary source control, or reads secret values.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from src.services.snapshot_paths import resolve_repo_root
_DEFAULT_SECURITY_DIR = resolve_repo_root(Path(__file__)) / "docs" / "security"
_SNAPSHOT_NAME = "github-target-private-backup-evidence-gate.snapshot.json"
_SCHEMA_VERSION = "github_target_private_backup_evidence_gate_v1"
def load_latest_github_target_private_backup_evidence_gate(
security_dir: Path | None = None,
) -> dict[str, Any]:
"""Load the committed GitHub private backup evidence gate snapshot."""
directory = security_dir or _DEFAULT_SECURITY_DIR
latest = directory / _SNAPSHOT_NAME
with latest.open(encoding="utf-8") as handle:
payload = json.load(handle)
if not isinstance(payload, dict):
raise ValueError(f"{latest}: expected JSON object")
_require_schema(payload, str(latest))
_require_summary_consistency(payload, str(latest))
_require_fail_closed_boundaries(payload, str(latest))
_require_target_gate_consistency(payload, str(latest))
_require_no_secret_payload_keys(payload, str(latest))
return payload
def _require_schema(payload: dict[str, Any], label: str) -> None:
actual = payload.get("schema_version")
if actual != _SCHEMA_VERSION:
raise ValueError(f"{label}: expected schema_version={_SCHEMA_VERSION}, got {actual!r}")
def _require_summary_consistency(payload: dict[str, Any], label: str) -> None:
summary = payload.get("summary") or {}
targets = payload.get("targets") or []
approval_targets = [target for target in targets if target.get("approval_required") is True]
public_visible = [
target
for target in approval_targets
if str(target.get("probe_status") or "").startswith("exists")
]
not_found_or_private = [
target
for target in approval_targets
if target.get("probe_status") == "not_found_or_private"
]
expected = {
"target_decision_count": len(targets),
"approval_required_target_count": len(approval_targets),
"public_probe_visible_target_count": len(public_visible),
"not_found_or_private_target_count": len(not_found_or_private),
"private_backup_verified_count": 0,
"private_visibility_evidence_missing_count": len(approval_targets),
"safe_credential_required_count": len(approval_targets),
"safe_credential_accepted_evidence_count": 0,
"execution_ready_count": 0,
"blocked_target_count": len(approval_targets),
}
mismatches = {
key: {"expected": value, "actual": summary.get(key)}
for key, value in expected.items()
if summary.get(key) != value
}
if mismatches:
raise ValueError(f"{label}: summary mismatch {mismatches}")
if summary.get("public_repo_allowed") is not False:
raise ValueError(f"{label}: public_repo_allowed must stay false")
if summary.get("not_found_or_private_as_absent_allowed") is not False:
raise ValueError(f"{label}: not_found_or_private_as_absent_allowed must stay false")
def _require_fail_closed_boundaries(payload: dict[str, Any], label: str) -> None:
boundaries = payload.get("operation_boundaries") or {}
if boundaries.get("read_only_api_allowed") is not True:
raise ValueError(f"{label}: read_only_api_allowed must be true")
blocked_operation_flags = {
"github_api_write_allowed",
"gitea_api_write_allowed",
"repo_creation_allowed",
"visibility_change_allowed",
"refs_sync_allowed",
"workflow_modification_allowed",
"workflow_trigger_allowed",
"github_primary_switch_allowed",
"secret_value_collection_allowed",
"private_clone_url_collection_allowed",
}
allowed_operations = sorted(
flag for flag in blocked_operation_flags if boundaries.get(flag) is not False
)
if allowed_operations:
raise ValueError(f"{label}: operation boundaries must remain false: {allowed_operations}")
flags = payload.get("authorization_flags") or {}
allowed_flags = sorted(flag for flag, value in flags.items() if value is not False)
if allowed_flags:
raise ValueError(f"{label}: authorization flags must remain false: {allowed_flags}")
summary = payload.get("summary") or {}
false_summary_flags = {
"repo_creation_authorized",
"visibility_change_authorized",
"refs_sync_authorized",
"github_primary_switch_authorized",
"workflow_modification_authorized",
"workflow_trigger_authorized",
"secret_value_collection_allowed",
"private_clone_url_collection_allowed",
}
allowed_summary = sorted(flag for flag in false_summary_flags if summary.get(flag) is not False)
if allowed_summary:
raise ValueError(f"{label}: summary authorization flags must remain false: {allowed_summary}")
def _require_target_gate_consistency(payload: dict[str, Any], label: str) -> None:
targets = payload.get("targets") or []
for target in targets:
if target.get("approval_required") is not True:
continue
repo = target.get("github_repo")
if target.get("private_backup_verified") is not False:
raise ValueError(f"{label}: {repo} private_backup_verified must stay false")
if target.get("refs_sync_ready") is not False or target.get("execution_ready") is not False:
raise ValueError(f"{label}: {repo} refs_sync_ready/execution_ready must stay false")
if not target.get("blockers"):
raise ValueError(f"{label}: {repo} must keep blockers until owner evidence is accepted")
false_flags = {
"repo_creation_authorized",
"visibility_change_authorized",
"refs_sync_authorized",
"github_primary_switch_authorized",
"secret_values_collected",
}
allowed = sorted(flag for flag in false_flags if target.get(flag) is not False)
if allowed:
raise ValueError(f"{label}: {repo} target flags must remain false: {allowed}")
status = str(target.get("visibility_evidence_status") or "")
probe_status = str(target.get("probe_status") or "")
if probe_status.startswith("exists") and "public_probe_visible" not in status:
raise ValueError(f"{label}: {repo} public probe visibility must be blocked")
if probe_status == "not_found_or_private" and "not_verified" not in status:
raise ValueError(f"{label}: {repo} not_found_or_private must not be private verification")
def _require_no_secret_payload_keys(payload: Any, label: str, path: str = "$") -> None:
forbidden_fragments = {
"token",
"secret",
"private_key",
"cookie",
"session",
"credential",
"authorization",
"clone_url",
}
if isinstance(payload, dict):
for key, value in payload.items():
normalized = str(key).lower()
if any(fragment in normalized for fragment in forbidden_fragments):
if normalized not in {
"secret_value_collection_allowed",
"secret_values_collected",
"private_clone_url_collection_allowed",
"safe_credential_evidence_status",
"safe_credential_evidence_ref",
"safe_credential_required_count",
"safe_credential_accepted_evidence_count",
"forbidden_action_count",
"authorization_flags",
}:
raise ValueError(f"{label}: forbidden secret payload key at {path}.{key}")
_require_no_secret_payload_keys(value, label, f"{path}.{key}")
elif isinstance(payload, list):
for index, value in enumerate(payload):
_require_no_secret_payload_keys(value, label, f"{path}[{index}]")

View File

@@ -0,0 +1,171 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
from src.services.github_target_private_backup_evidence_gate import (
load_latest_github_target_private_backup_evidence_gate,
)
def test_load_latest_github_target_private_backup_evidence_gate_reads_committed_snapshot():
data = load_latest_github_target_private_backup_evidence_gate()
assert data["schema_version"] == "github_target_private_backup_evidence_gate_v1"
assert data["status"] == "blocked_public_visibility_and_safe_credential_evidence_required"
assert data["summary"]["approval_required_target_count"] == 9
assert data["summary"]["public_probe_visible_target_count"] == 4
assert data["summary"]["not_found_or_private_target_count"] == 5
assert data["summary"]["private_backup_verified_count"] == 0
assert data["summary"]["safe_credential_accepted_evidence_count"] == 0
assert data["summary"]["safe_credential_required_count"] == 9
assert data["summary"]["refs_sync_authorized"] is False
assert data["summary"]["public_repo_allowed"] is False
assert data["operation_boundaries"]["read_only_api_allowed"] is True
assert data["operation_boundaries"]["repo_creation_allowed"] is False
assert data["authorization_flags"]["repo_creation_authorized"] is False
assert data["authorization_flags"]["secret_values_collected"] is False
def test_private_backup_gate_rejects_runtime_authorization(tmp_path: Path):
snapshot = _snapshot()
snapshot["authorization_flags"]["refs_sync_authorized"] = True
_write_snapshot(tmp_path, snapshot)
with pytest.raises(ValueError, match="authorization flags"):
load_latest_github_target_private_backup_evidence_gate(tmp_path)
def test_private_backup_gate_rejects_summary_drift(tmp_path: Path):
snapshot = _snapshot()
snapshot["summary"]["private_backup_verified_count"] = 1
_write_snapshot(tmp_path, snapshot)
with pytest.raises(ValueError, match="summary mismatch"):
load_latest_github_target_private_backup_evidence_gate(tmp_path)
def test_private_backup_gate_rejects_public_target_marked_private(tmp_path: Path):
snapshot = _snapshot()
snapshot["targets"][0]["private_backup_verified"] = True
_write_snapshot(tmp_path, snapshot)
with pytest.raises(ValueError, match="private_backup_verified"):
load_latest_github_target_private_backup_evidence_gate(tmp_path)
def test_private_backup_gate_rejects_not_found_as_verified(tmp_path: Path):
snapshot = _snapshot()
snapshot["targets"][1]["visibility_evidence_status"] = "private_verified"
_write_snapshot(tmp_path, snapshot)
with pytest.raises(ValueError, match="not_found_or_private"):
load_latest_github_target_private_backup_evidence_gate(tmp_path)
def _write_snapshot(directory: Path, snapshot: dict) -> None:
(directory / "github-target-private-backup-evidence-gate.snapshot.json").write_text(
json.dumps(snapshot),
encoding="utf-8",
)
def _snapshot() -> dict:
return {
"schema_version": "github_target_private_backup_evidence_gate_v1",
"generated_at": "2026-06-26T00:00:00+00:00",
"status": "blocked_public_visibility_and_safe_credential_evidence_required",
"mode": "read_only_private_backup_evidence_gate",
"source_reviews": {},
"summary": {
"target_decision_count": 3,
"approval_required_target_count": 2,
"approval_package_item_count": 2,
"public_probe_visible_target_count": 1,
"not_found_or_private_target_count": 1,
"private_backup_verified_count": 0,
"private_visibility_evidence_missing_count": 2,
"safe_credential_required_count": 2,
"safe_credential_accepted_evidence_count": 0,
"owner_response_received_count": 0,
"owner_response_accepted_count": 0,
"execution_ready_count": 0,
"blocked_target_count": 2,
"external_scope_target_count": 1,
"forbidden_action_count": 12,
"repo_creation_authorized": False,
"visibility_change_authorized": False,
"refs_sync_authorized": False,
"github_primary_switch_authorized": False,
"workflow_modification_authorized": False,
"workflow_trigger_authorized": False,
"secret_value_collection_allowed": False,
"private_clone_url_collection_allowed": False,
"not_found_or_private_as_absent_allowed": False,
"public_repo_allowed": False,
},
"targets": [
_target("owenhytsai/awoooi", True, "exists", "blocked_public_probe_visible_private_evidence_required"),
_target("owenhytsai/VibeWork", True, "not_found_or_private", "blocked_private_or_absent_not_verified"),
_target("nexu-io/open-design", False, "exists", "external_scope_not_backup_target"),
],
"acceptance_requirements": ["private evidence required"],
"rejection_rules": ["reject secrets"],
"operation_boundaries": {
"read_only_api_allowed": True,
"github_api_write_allowed": False,
"gitea_api_write_allowed": False,
"repo_creation_allowed": False,
"visibility_change_allowed": False,
"refs_sync_allowed": False,
"workflow_modification_allowed": False,
"workflow_trigger_allowed": False,
"github_primary_switch_allowed": False,
"secret_value_collection_allowed": False,
"private_clone_url_collection_allowed": False,
},
"authorization_flags": {
"runtime_execution_authorized": False,
"repo_creation_authorized": False,
"visibility_change_authorized": False,
"refs_sync_authorized": False,
"workflow_modification_authorized": False,
"workflow_trigger_authorized": False,
"github_primary_switch_authorized": False,
"secret_values_collected": False,
},
}
def _target(
repo: str,
approval_required: bool,
probe_status: str,
visibility_status: str,
) -> dict:
return {
"github_repo": repo,
"source_key": repo,
"approval_required": approval_required,
"probe_status": probe_status,
"target_state": probe_status,
"risk": "HIGH",
"visibility_evidence_status": visibility_status,
"private_backup_verified": False,
"private_visibility_owner_evidence_ref": None,
"safe_credential_evidence_status": "missing_safe_credential_metadata",
"safe_credential_evidence_ref": None,
"owner_response_accepted": False,
"refs_sync_ready": False,
"execution_ready": False,
"blockers": ["blocked"],
"evidence_refs": ["docs/security/github-target-decision.snapshot.json"],
"forbidden_actions": ["push_refs"],
"repo_creation_authorized": False,
"visibility_change_authorized": False,
"refs_sync_authorized": False,
"github_primary_switch_authorized": False,
"secret_values_collected": False,
}

View File

@@ -0,0 +1,44 @@
from __future__ import annotations
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.api.v1.agents import router
def test_github_target_private_backup_evidence_gate_endpoint_returns_committed_snapshot():
app = FastAPI()
app.include_router(router, prefix="/api/v1")
client = TestClient(app)
response = client.get("/api/v1/agents/github-target-private-backup-evidence-gate")
assert response.status_code == 200
data = response.json()
assert data["schema_version"] == "github_target_private_backup_evidence_gate_v1"
assert data["status"] == "blocked_public_visibility_and_safe_credential_evidence_required"
assert data["summary"]["approval_required_target_count"] == 9
assert data["summary"]["public_probe_visible_target_count"] == 4
assert data["summary"]["not_found_or_private_target_count"] == 5
assert data["summary"]["private_backup_verified_count"] == 0
assert data["summary"]["safe_credential_accepted_evidence_count"] == 0
assert data["summary"]["execution_ready_count"] == 0
assert data["summary"]["public_repo_allowed"] is False
assert data["operation_boundaries"]["read_only_api_allowed"] is True
assert data["operation_boundaries"]["repo_creation_allowed"] is False
assert data["operation_boundaries"]["refs_sync_allowed"] is False
assert data["operation_boundaries"]["workflow_trigger_allowed"] is False
assert data["authorization_flags"]["runtime_execution_authorized"] is False
assert data["authorization_flags"]["repo_creation_authorized"] is False
assert data["authorization_flags"]["secret_values_collected"] is False
public_target = next(target for target in data["targets"] if target["github_repo"] == "owenhytsai/awoooi")
assert public_target["visibility_evidence_status"] == (
"blocked_public_probe_visible_private_evidence_required"
)
private_or_absent_target = next(
target for target in data["targets"] if target["github_repo"] == "owenhytsai/VibeWork"
)
assert private_or_absent_target["visibility_evidence_status"] == (
"blocked_private_or_absent_not_verified"
)