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

feat(github): add safe credential intake readiness

Browse Source
This commit is contained in:
Your Name
2026-06-27 21:48:33 +08:00
parent 1a6f8f4275
commit f917ea41c2
4 changed files with 363 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -93,6 +93,14 @@ def build_github_target_private_backup_evidence_gate(
}
owner_summary = _dict(owner_response.get("summary"))
owner_response_intake_readiness = _owner_response_intake_readiness(owner_response)
safe_credential_evidence_intake_readiness = (
_safe_credential_evidence_intake_readiness(
owner_response=owner_response,
required_target_count=len(
[row for row in decisions if row.get("approval_required") is True]
),
)
)
owner_template_by_repo = {
str(row.get("github_repo")): _dict(row)
for row in _list(owner_response.get("response_templates"))
@@ -214,6 +222,25 @@ def build_github_target_private_backup_evidence_gate(
- private_backup_verified_count,
"safe_credential_required_count": len(approval_required_targets),
"safe_credential_accepted_evidence_count": 0,
"safe_credential_evidence_intake_ready": safe_credential_evidence_intake_readiness[
"intake_ready"
],
"safe_credential_required_redacted_evidence_ref_count": safe_credential_evidence_intake_readiness[
"required_redacted_evidence_ref_count"
],
"safe_credential_evidence_ref_rule_count": safe_credential_evidence_intake_readiness[
"evidence_ref_rule_count"
],
"safe_credential_redaction_example_count": safe_credential_evidence_intake_readiness[
"redaction_example_count"
],
"safe_credential_forbidden_payload_count": safe_credential_evidence_intake_readiness[
"forbidden_payload_count"
],
"safe_credential_quarantine_lane_count": safe_credential_evidence_intake_readiness[
"quarantine_lane_count"
],
"safe_credential_raw_payload_storage_allowed": False,
"owner_response_received_count": _int(
owner_summary.get("received_response_count")
),
@@ -264,6 +291,7 @@ def build_github_target_private_backup_evidence_gate(
"public_repo_allowed": False,
},
"owner_response_intake_readiness": owner_response_intake_readiness,
"safe_credential_evidence_intake_readiness": safe_credential_evidence_intake_readiness,
"targets": targets,
"acceptance_requirements": _acceptance_requirements(owner_response),
"rejection_rules": _rejection_rules(owner_response),
@@ -367,6 +395,21 @@ def _build_target(
"private_visibility_owner_evidence_ref": connector_readback.get("evidence_ref"),
"safe_credential_evidence_status": "not_collected",
"safe_credential_evidence_ref": None,
"safe_credential_evidence_intake_ready": approval_required,
"safe_credential_evidence_submission_status": "waiting_redacted_evidence_ref"
if approval_required
else "not_required",
"safe_credential_required_redacted_evidence_ref": approval_required,
"safe_credential_allowed_evidence_ref_types": [
"repo_path",
"snapshot_path",
"redacted_metadata_pointer",
]
if approval_required
else [],
"safe_credential_raw_payload_storage_allowed": False,
"safe_credential_private_clone_url_collection_allowed": False,
"safe_credential_secret_value_collection_allowed": False,
"owner_response_template_id": owner_response_template.get("template_id"),
"owner_response_submission_status": "waiting_owner_response"
if approval_required
@@ -687,6 +730,9 @@ def _require_owner_response_intake_consistency(
_dict(row) for row in _list(payload.get("intake_preflight_checks"))
]
acceptance_checks = [_dict(row) for row in _list(payload.get("acceptance_checks"))]
redaction_examples = [
_dict(row) for row in _list(payload.get("owner_response_redaction_examples"))
]
summary = _dict(payload.get("summary"))
if (
@@ -712,6 +758,10 @@ def _require_owner_response_intake_consistency(
raise ValueError(f"{label}: intake preflight check count must match checks")
if _int(summary.get("acceptance_check_count")) != len(acceptance_checks):
raise ValueError(f"{label}: acceptance check count must match checks")
if _int(summary.get("owner_response_redaction_example_count")) != len(
redaction_examples
):
raise ValueError(f"{label}: redaction example count must match examples")
template_ids = {
str(row.get("template_id")) for row in templates if row.get("template_id")
@@ -723,12 +773,18 @@ def _require_owner_response_intake_consistency(
)
executable_rows = [
str(row.get("template_id") or row.get("check_id") or "unknown")
str(
row.get("template_id")
or row.get("check_id")
or row.get("example_id")
or "unknown"
)
for row in [
*templates,
*collection_checks,
*preflight_checks,
*acceptance_checks,
*redaction_examples,
]
if row.get("execution_authorized") is not False
]
@@ -737,6 +793,31 @@ def _require_owner_response_intake_consistency(
f"{label}: owner response intake rows must remain non-executing: {executable_rows}"
)
raw_payload_examples = [
str(row.get("example_id") or "unknown")
for row in redaction_examples
if row.get("stored_raw_payload_allowed") is not False
]
if raw_payload_examples:
raise ValueError(
f"{label}: redaction examples must not allow raw payload storage: {raw_payload_examples}"
)
required_forbidden_payloads = {
"token_value",
"secret_value",
"private_clone_url_credential",
"repo_archive",
"git_object_pack",
}
missing_forbidden_payloads = sorted(
required_forbidden_payloads - set(_strings(packet.get("forbidden_payloads")))
)
if missing_forbidden_payloads:
raise ValueError(
f"{label}: forbidden payloads missing safe credential blockers: {missing_forbidden_payloads}"
)
def _owner_response_intake_readiness(owner_response: dict[str, Any]) -> dict[str, Any]:
summary = _dict(owner_response.get("summary"))
@@ -805,6 +886,104 @@ def _owner_response_intake_readiness(owner_response: dict[str, Any]) -> dict[str
}
def _safe_credential_evidence_intake_readiness(
*,
owner_response: dict[str, Any],
required_target_count: int,
) -> dict[str, Any]:
summary = _dict(owner_response.get("summary"))
packet = _dict(owner_response.get("owner_response_request_packet"))
evidence_ref_rules = _strings(packet.get("evidence_ref_rules"))
redaction_examples = [
_redaction_example_summary(row)
for row in _list(owner_response.get("owner_response_redaction_examples"))
]
forbidden_payloads = _strings(packet.get("forbidden_payloads"))
checks = [
*[
_dict(row)
for row in _list(owner_response.get("owner_response_collection_checks"))
],
*[_dict(row) for row in _list(owner_response.get("intake_preflight_checks"))],
*[_dict(row) for row in _list(owner_response.get("acceptance_checks"))],
]
quarantine_lanes = sorted(
{
str(row.get("failure_lane"))
for row in checks
if str(row.get("failure_lane") or "").startswith("quarantine")
}
)
accepted_evidence_count = 0
required_redacted_evidence_ref_count = (
required_target_count - accepted_evidence_count
)
required_forbidden_payloads = {
"token_value",
"secret_value",
"private_clone_url_credential",
"repo_archive",
"git_object_pack",
}
intake_ready = (
packet.get("execution_authorized") is False
and packet.get("not_approval") is True
and bool(evidence_ref_rules)
and bool(redaction_examples)
and _int(summary.get("owner_response_redaction_example_count"))
== len(redaction_examples)
and required_forbidden_payloads.issubset(set(forbidden_payloads))
)
return {
"status": "ready_to_collect_redacted_evidence_refs_not_credentials"
if intake_ready
else "blocked_safe_credential_evidence_intake_contract_incomplete",
"intake_ready": intake_ready,
"required_target_count": required_target_count,
"accepted_evidence_count": accepted_evidence_count,
"required_redacted_evidence_ref_count": required_redacted_evidence_ref_count,
"allowed_evidence_ref_types": [
"repo_path",
"snapshot_path",
"redacted_metadata_pointer",
],
"evidence_ref_rule_count": len(evidence_ref_rules),
"evidence_ref_rules": evidence_ref_rules,
"redaction_example_count": len(redaction_examples),
"redaction_examples": redaction_examples,
"forbidden_payload_count": len(forbidden_payloads),
"forbidden_payloads": forbidden_payloads,
"quarantine_lane_count": len(quarantine_lanes),
"quarantine_lanes": quarantine_lanes,
"allowed_submission_modes": _strings(packet.get("allowed_submission_modes")),
"stored_raw_payload_allowed": False,
"secret_value_collection_allowed": False,
"private_clone_url_collection_allowed": False,
"credential_value_collection_allowed": False,
"target_owner_request_dispatch_authorized": False,
"action_buttons_allowed": False,
"execution_authorized": False,
"not_approval": True,
}
def _redaction_example_summary(value: Any) -> dict[str, Any]:
row = _dict(value)
return {
"example_id": str(row.get("example_id") or ""),
"display_order": _int(row.get("display_order")),
"example_status": str(row.get("example_status") or ""),
"category": str(row.get("category") or ""),
"safe_response_shape": _strings(row.get("safe_response_shape")),
"required_redactions": _strings(row.get("required_redactions")),
"forbidden_raw_values": _strings(row.get("forbidden_raw_values")),
"stored_raw_payload_allowed": False,
"execution_authorized": False,
"not_approval": row.get("not_approval") is True,
}
def _owner_response_template_summary(value: Any) -> dict[str, Any]:
row = _dict(value)
return {
@@ -888,6 +1067,6 @@ def _strings(value: Any) -> list[str]:
def _int(value: Any) -> int:
if isinstance(value, bool):
return int(value)
if isinstance(value, (int, float)):
if isinstance(value, int | float):
return int(value)
return 0