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

feat(github): classify missing private backup targets

Browse Source
This commit is contained in:
Your Name
2026-06-27 19:32:58 +08:00
parent d359587316
commit 31215e5d11
4 changed files with 291 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -21,6 +21,7 @@ _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"
_MISSING_SOURCE_READINESS_FILE = "github-target-missing-source-readiness.snapshot.json"
def load_latest_github_target_private_backup_evidence_gate(
@@ -33,6 +34,7 @@ def load_latest_github_target_private_backup_evidence_gate(
approval_package = _load_snapshot(directory / _APPROVAL_PACKAGE_FILE)
probe = _load_snapshot(directory / _PROBE_FILE)
connector_readback = _load_optional_snapshot(directory / _CONNECTOR_READBACK_FILE)
missing_source_readiness = _load_optional_snapshot(directory / _MISSING_SOURCE_READINESS_FILE)
_require_source_contracts(
decision=decision,
@@ -40,6 +42,7 @@ def load_latest_github_target_private_backup_evidence_gate(
approval_package=approval_package,
probe=probe,
connector_readback=connector_readback,
missing_source_readiness=missing_source_readiness,
)
return build_github_target_private_backup_evidence_gate(
decision=decision,
@@ -47,6 +50,7 @@ def load_latest_github_target_private_backup_evidence_gate(
approval_package=approval_package,
probe=probe,
connector_readback=connector_readback,
missing_source_readiness=missing_source_readiness,
)
@@ -57,10 +61,13 @@ def build_github_target_private_backup_evidence_gate(
approval_package: dict[str, Any],
probe: dict[str, Any],
connector_readback: dict[str, Any] | None = None,
missing_source_readiness: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build the read-only gate response from source-control snapshots."""
connector_payload = _dict(connector_readback)
owner_readback = _dict(connector_payload.get("owner_readback"))
missing_payload = _dict(missing_source_readiness)
missing_summary = _dict(missing_payload.get("summary"))
decisions = [_dict(row) for row in _list(decision.get("decisions"))]
probe_by_repo = {
str(row.get("github_repo")): _dict(row)
@@ -77,6 +84,11 @@ def build_github_target_private_backup_evidence_gate(
for row in _list(connector_payload.get("targets"))
if row.get("github_repo")
}
missing_source_by_repo = {
str(row.get("github_repo")): _dict(row)
for row in _list(missing_payload.get("targets"))
if row.get("github_repo")
}
owner_summary = _dict(owner_response.get("summary"))
targets = [
@@ -85,6 +97,7 @@ def build_github_target_private_backup_evidence_gate(
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")), {}),
missing_source_readiness=missing_source_by_repo.get(str(row.get("github_repo")), {}),
)
for row in decisions
]
@@ -125,6 +138,7 @@ def build_github_target_private_backup_evidence_gate(
"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}",
"github_missing_source_readiness": f"docs/security/{_MISSING_SOURCE_READINESS_FILE}",
},
"summary": {
"target_decision_count": len(targets),
@@ -147,6 +161,23 @@ def build_github_target_private_backup_evidence_gate(
for row in _list(connector_payload.get("targets"))
if _dict(row).get("readback_status") == "not_found_or_inaccessible"
),
"github_missing_target_resolution_count": len(_list(missing_payload.get("targets"))),
"github_missing_target_gitea_source_candidate_count": _int(
missing_summary.get("gitea_source_candidate_count")
),
"github_missing_target_internal_remote_source_candidate_count": _int(
missing_summary.get("internal_remote_source_candidate_count")
),
"github_missing_target_local_worktree_candidate_count": _int(
missing_summary.get("local_worktree_candidate_count")
),
"github_missing_target_canonical_source_ambiguous_count": _int(
missing_summary.get("canonical_source_ambiguous_count")
),
"github_missing_target_create_private_repo_ready_count": _int(
missing_summary.get("create_private_repo_ready_count")
),
"github_missing_target_refs_sync_ready_count": _int(missing_summary.get("refs_sync_ready_count")),
"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,
@@ -206,6 +237,7 @@ def _build_target(
probe: dict[str, Any],
approval_item: dict[str, Any],
connector_readback: dict[str, Any],
missing_source_readiness: 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")
@@ -245,6 +277,15 @@ def _build_target(
"visibility_evidence_status": visibility_status,
"github_connector_readback_status": str(connector_readback.get("readback_status") or "not_checked"),
"github_connector_visibility": connector_readback.get("visibility"),
"missing_target_source_resolution_status": missing_source_readiness.get("source_resolution_status"),
"missing_target_source_candidate_type": missing_source_readiness.get("source_candidate_type"),
"missing_target_gitea_repo": missing_source_readiness.get("gitea_repo"),
"missing_target_local_worktree_status": missing_source_readiness.get("local_worktree_status"),
"missing_target_create_private_repo_ready": missing_source_readiness.get("create_private_repo_ready"),
"missing_target_refs_sync_ready": missing_source_readiness.get("refs_sync_ready"),
"missing_target_required_owner_decisions": _strings(
missing_source_readiness.get("required_owner_decisions")
),
"private_backup_verified": private_visibility_verified,
"private_visibility_owner_evidence_ref": connector_readback.get("evidence_ref"),
"safe_credential_evidence_status": "not_collected",
@@ -323,6 +364,7 @@ def _require_source_contracts(
approval_package: dict[str, Any],
probe: dict[str, Any],
connector_readback: dict[str, Any],
missing_source_readiness: 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)
@@ -330,11 +372,19 @@ def _require_source_contracts(
_require_schema(probe, "github_target_probe_v1", _PROBE_FILE)
if connector_readback:
_require_schema(connector_readback, "github_target_connector_readback_v1", _CONNECTOR_READBACK_FILE)
if missing_source_readiness:
_require_schema(
missing_source_readiness,
"github_target_missing_source_readiness_v1",
_MISSING_SOURCE_READINESS_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)
if missing_source_readiness:
_require_missing_source_readiness_consistency(missing_source_readiness, _MISSING_SOURCE_READINESS_FILE)
_require_owner_response_boundaries(owner_response, _OWNER_RESPONSE_FILE)
@@ -402,6 +452,42 @@ def _require_connector_readback_consistency(payload: dict[str, Any], label: str)
raise ValueError(f"{label}: connector readback must remain read-only: {enabled}")
def _require_missing_source_readiness_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")
create_ready = sum(1 for row in targets if row.get("create_private_repo_ready") is True)
refs_ready = sum(1 for row in targets if row.get("refs_sync_ready") is True)
owner_required = sum(1 for row in targets if row.get("owner_response_required") is True)
if _int(summary.get("create_private_repo_ready_count")) != create_ready:
raise ValueError(f"{label}: create ready count must match targets")
if _int(summary.get("refs_sync_ready_count")) != refs_ready:
raise ValueError(f"{label}: refs sync ready count must match targets")
if _int(summary.get("owner_response_required_count")) != owner_required:
raise ValueError(f"{label}: owner response required count must match targets")
false_summary_flags = {
"github_api_write_performed",
"repo_creation_performed",
"visibility_change_performed",
"refs_sync_performed",
"secret_values_collected",
}
enabled = sorted(flag for flag in false_summary_flags if summary.get(flag) is not False)
boundaries = _dict(payload.get("operation_boundaries"))
false_boundary_flags = {
"github_api_write_allowed",
"repo_creation_allowed",
"visibility_change_allowed",
"refs_sync_allowed",
"secret_value_collection_allowed",
}
enabled.extend(sorted(flag for flag in false_boundary_flags if boundaries.get(flag) is not False))
if enabled:
raise ValueError(f"{label}: missing source readiness 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")