wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(github): read back private backup visibility evidence

Browse Source
This commit is contained in:
Your Name
2026-06-27 11:32:05 +08:00
parent 8fdcc0194f
commit c78b19d1dc
5 changed files with 213 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
apps/api/src/services/github_target_private_backup_evidence_gate.py
View File

@@ -20,6 +20,7 @@ _DECISION_FILE = "github-target-decision.snapshot.json"
_OWNER_RESPONSE_FILE = "github-target-owner-decision-response.snapshot.json"
_APPROVAL_PACKAGE_FILE = "github-target-repo-approval-package.snapshot.json"
_PROBE_FILE = "github-target-probe.snapshot.json"
_CONNECTOR_READBACK_FILE = "github-target-connector-readback.snapshot.json"
def load_latest_github_target_private_backup_evidence_gate(
@@ -31,18 +32,21 @@ def load_latest_github_target_private_backup_evidence_gate(
owner_response = _load_snapshot(directory / _OWNER_RESPONSE_FILE)
approval_package = _load_snapshot(directory / _APPROVAL_PACKAGE_FILE)
probe = _load_snapshot(directory / _PROBE_FILE)
connector_readback = _load_optional_snapshot(directory / _CONNECTOR_READBACK_FILE)
_require_source_contracts(
decision=decision,
owner_response=owner_response,
approval_package=approval_package,
probe=probe,
connector_readback=connector_readback,
)
return build_github_target_private_backup_evidence_gate(
decision=decision,
owner_response=owner_response,
approval_package=approval_package,
probe=probe,
connector_readback=connector_readback,
)
@@ -52,8 +56,10 @@ def build_github_target_private_backup_evidence_gate(
owner_response: dict[str, Any],
approval_package: dict[str, Any],
probe: dict[str, Any],
connector_readback: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build the read-only gate response from source-control snapshots."""
connector_payload = _dict(connector_readback)
decisions = [_dict(row) for row in _list(decision.get("decisions"))]
probe_by_repo = {
str(row.get("github_repo")): _dict(row)
@@ -65,6 +71,11 @@ def build_github_target_private_backup_evidence_gate(
for row in _list(approval_package.get("approval_items"))
if row.get("github_repo")
}
connector_by_repo = {
str(row.get("github_repo")): _dict(row)
for row in _list(connector_payload.get("targets"))
if row.get("github_repo")
}
owner_summary = _dict(owner_response.get("summary"))
targets = [
@@ -72,6 +83,7 @@ def build_github_target_private_backup_evidence_gate(
decision=row,
probe=probe_by_repo.get(str(row.get("github_repo")), {}),
approval_item=package_by_repo.get(str(row.get("github_repo")), {}),
connector_readback=connector_by_repo.get(str(row.get("github_repo")), {}),
)
for row in decisions
]
@@ -111,14 +123,27 @@ def build_github_target_private_backup_evidence_gate(
"owner_decision_response": f"docs/security/{_OWNER_RESPONSE_FILE}",
"approval_package": f"docs/security/{_APPROVAL_PACKAGE_FILE}",
"github_target_probe": f"docs/security/{_PROBE_FILE}",
"github_connector_readback": f"docs/security/{_CONNECTOR_READBACK_FILE}",
},
"summary": {
"target_decision_count": len(targets),
"approval_required_target_count": len(approval_required_targets),
"approval_package_item_count": len(_list(approval_package.get("approval_items"))),
"github_connector_readback_count": len(_list(connector_payload.get("targets"))),
"github_connector_private_visibility_count": sum(
1
for row in _list(connector_payload.get("targets"))
if _dict(row).get("private_visibility_verified") is True
),
"github_connector_not_found_or_inaccessible_count": sum(
1
for row in _list(connector_payload.get("targets"))
if _dict(row).get("readback_status") == "not_found_or_inaccessible"
),
"public_probe_visible_target_count": len(public_probe_visible_targets),
"not_found_or_private_target_count": len(not_found_or_private_targets),
"private_backup_verified_count": private_backup_verified_count,
"private_visibility_verified_count": private_backup_verified_count,
"private_visibility_evidence_missing_count": len(approval_required_targets) - private_backup_verified_count,
"safe_credential_required_count": len(approval_required_targets),
"safe_credential_accepted_evidence_count": 0,
@@ -173,17 +198,25 @@ def _build_target(
decision: dict[str, Any],
probe: dict[str, Any],
approval_item: dict[str, Any],
connector_readback: dict[str, Any],
) -> dict[str, Any]:
github_repo = str(decision.get("github_repo") or "")
probe_status = str(probe.get("status") or decision.get("probe_status") or "unknown")
target_state = str(decision.get("target_state") or "unknown")
approval_required = decision.get("approval_required") is True
external_scope = not approval_required or target_state == "external_scope"
private_visibility_verified = (
not external_scope
and connector_readback.get("readback_status") == "visible"
and connector_readback.get("visibility") == "private"
and connector_readback.get("private_visibility_verified") is True
)
visibility_status = _visibility_evidence_status(
external_scope=external_scope,
probe_status=probe_status,
private_visibility_verified=private_visibility_verified,
)
blockers = _target_blockers(visibility_status, approval_required)
blockers = _target_blockers(visibility_status, approval_required, private_visibility_verified)
forbidden_actions = _strings(approval_item.get("still_forbidden")) or [
"create_github_repo",
"change_repo_visibility",
@@ -203,8 +236,10 @@ def _build_target(
"target_state": target_state,
"risk": str(decision.get("risk") or "UNKNOWN"),
"visibility_evidence_status": visibility_status,
"private_backup_verified": False,
"private_visibility_owner_evidence_ref": None,
"github_connector_readback_status": str(connector_readback.get("readback_status") or "not_checked"),
"github_connector_visibility": connector_readback.get("visibility"),
"private_backup_verified": private_visibility_verified,
"private_visibility_owner_evidence_ref": connector_readback.get("evidence_ref"),
"safe_credential_evidence_status": "not_collected",
"safe_credential_evidence_ref": None,
"owner_response_accepted": False,
@@ -222,9 +257,16 @@ def _build_target(
}
def _visibility_evidence_status(*, external_scope: bool, probe_status: str) -> str:
def _visibility_evidence_status(
*,
external_scope: bool,
probe_status: str,
private_visibility_verified: bool,
) -> str:
if external_scope:
return "external_scope_not_backup_target"
if private_visibility_verified:
return "verified_private_by_github_connector_readback"
if probe_status == "exists":
return "blocked_public_probe_visible_private_evidence_required"
if probe_status == "not_found_or_private":
@@ -232,15 +274,20 @@ def _visibility_evidence_status(*, external_scope: bool, probe_status: str) -> s
return "blocked_probe_status_unknown"
def _target_blockers(visibility_status: str, approval_required: bool) -> list[str]:
def _target_blockers(
visibility_status: str,
approval_required: bool,
private_visibility_verified: bool,
) -> list[str]:
if not approval_required:
return ["external_scope_not_backup_target"]
blockers = [
"private_visibility_owner_evidence_missing",
"safe_credential_evidence_missing",
"owner_response_not_accepted",
"refs_sync_not_authorized",
]
if not private_visibility_verified:
blockers.insert(0, "private_visibility_owner_evidence_missing")
if visibility_status == "blocked_public_probe_visible_private_evidence_required":
blockers.insert(0, "public_probe_visible_not_private_backup")
if visibility_status == "blocked_private_or_absent_not_verified":
@@ -256,20 +303,31 @@ def _load_snapshot(path: Path) -> dict[str, Any]:
return payload
def _load_optional_snapshot(path: Path) -> dict[str, Any]:
if not path.exists():
return {}
return _load_snapshot(path)
def _require_source_contracts(
*,
decision: dict[str, Any],
owner_response: dict[str, Any],
approval_package: dict[str, Any],
probe: dict[str, Any],
connector_readback: dict[str, Any],
) -> None:
_require_schema(decision, "github_target_decision_v1", _DECISION_FILE)
_require_schema(owner_response, "github_target_owner_decision_response_v1", _OWNER_RESPONSE_FILE)
_require_schema(approval_package, "github_target_repo_approval_package_v1", _APPROVAL_PACKAGE_FILE)
_require_schema(probe, "github_target_probe_v1", _PROBE_FILE)
if connector_readback:
_require_schema(connector_readback, "github_target_connector_readback_v1", _CONNECTOR_READBACK_FILE)
_require_decision_consistency(decision, _DECISION_FILE)
_require_probe_consistency(probe, _PROBE_FILE)
_require_approval_package_consistency(approval_package, _APPROVAL_PACKAGE_FILE)
if connector_readback:
_require_connector_readback_consistency(connector_readback, _CONNECTOR_READBACK_FILE)
_require_owner_response_boundaries(owner_response, _OWNER_RESPONSE_FILE)
@@ -306,6 +364,29 @@ def _require_approval_package_consistency(payload: dict[str, Any], label: str) -
raise ValueError(f"{label}: package_count must equal approval_items length")
def _require_connector_readback_consistency(payload: dict[str, Any], label: str) -> None:
summary = _dict(payload.get("summary"))
targets = [_dict(row) for row in _list(payload.get("targets"))]
if _int(summary.get("target_count")) != len(targets):
raise ValueError(f"{label}: summary.target_count must equal targets length")
private_count = sum(1 for row in targets if row.get("private_visibility_verified") is True)
not_found_count = sum(1 for row in targets if row.get("readback_status") == "not_found_or_inaccessible")
if _int(summary.get("private_visibility_verified_count")) != private_count:
raise ValueError(f"{label}: private visibility count must match targets")
if _int(summary.get("not_found_or_inaccessible_count")) != not_found_count:
raise ValueError(f"{label}: not found count must match targets")
false_flags = {
"github_api_write_performed",
"repo_creation_performed",
"visibility_change_performed",
"refs_sync_performed",
"secret_values_collected",
}
enabled = sorted(flag for flag in false_flags if summary.get(flag) is not False)
if enabled:
raise ValueError(f"{label}: connector readback must remain read-only: {enabled}")
def _require_owner_response_boundaries(payload: dict[str, Any], label: str) -> None:
if payload.get("runtime_execution_authorized") is not False:
raise ValueError(f"{label}: runtime_execution_authorized must be false")