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

fix(gitea): close controlled runner attestation readback
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 1m6s
Details
CD Pipeline / build-and-deploy (push) Successful in 4m29s
Details
CD Pipeline / post-deploy-checks (push) Successful in 2m3s
Details

Browse Source
This commit is contained in:
ogt
2026-07-09 23:52:46 +08:00
parent c19a2cbc0e
commit 4a46252432
3 changed files with 77 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
apps/api/src/services/gitea_workflow_runner_owner_attestation_request.py
View File

@@ -14,7 +14,11 @@ from src.services.gitea_workflow_runner_health import (
)
_SCHEMA_VERSION = "gitea_workflow_runner_owner_attestation_request_v1"
_CONTRACT_ID = "ubuntu_latest_gitea_runner_label"
_LEGACY_CONTRACT_ID = "ubuntu_latest_gitea_runner_label"
_CONTROLLED_CONTRACT_IDS = (
"awoooi_non110_host_runner",
"awoooi_non110_ubuntu_runner",
)
_REQUIRED_FIELDS = [
"runner_label",
"actual_runner_mapping_ref",
@@ -53,47 +57,71 @@ def load_latest_gitea_workflow_runner_owner_attestation_request() -> dict[str, A
and contract.get("contract_id")
in set(_strings(rollups.get("runner_contracts_requiring_action")))
]
target_contract = next(
legacy_contract = next(
(
contract
for contract in action_contracts
if contract.get("contract_id") == _CONTRACT_ID
if contract.get("contract_id") == _LEGACY_CONTRACT_ID
),
{},
)
controlled_contracts = [
contract
for contract in health.get("runner_contracts", [])
if isinstance(contract, dict)
and contract.get("contract_id") in set(_CONTROLLED_CONTRACT_IDS)
]
target_contract = legacy_contract or _first_dict(controlled_contracts)
workflow_ids = _strings(rollups.get("workflow_ids_requiring_runner_attestation"))
request_ready = bool(target_contract) and bool(workflow_ids)
attestation_required = bool(legacy_contract) and bool(workflow_ids)
attestation_not_required = (
bool(controlled_contracts) and not action_contracts and not workflow_ids
)
request_packet_ready = attestation_required or attestation_not_required
active_blockers = []
if not request_ready:
if not request_packet_ready:
active_blockers.append("runner_attestation_request_source_missing")
active_blockers.append("owner_attestation_response_not_received")
if attestation_required:
active_blockers.append("owner_attestation_response_not_received")
status = "blocked_owner_attestation_request_source_missing"
if attestation_required:
status = "owner_attestation_request_ready_waiting_response"
elif attestation_not_required:
status = "owner_attestation_not_required_controlled_runner_contracts_mapped"
return {
"schema_version": _SCHEMA_VERSION,
"priority": "P1-002",
"scope": "gitea_workflow_runner_owner_attestation_request",
"status": (
"owner_attestation_request_ready_waiting_response"
if request_ready
else "blocked_owner_attestation_request_source_missing"
),
"status": status,
"readback": {
"workplan_id": "P1-002-GITEA-RUNNER-OWNER-ATTESTATION",
"source_health_schema_version": health.get("schema_version"),
"source_health_endpoint": "/api/v1/agents/gitea-workflow-runner-health",
"contract_id": _CONTRACT_ID,
"contract_id": str(target_contract.get("contract_id") or ""),
"runner_label": _first_string(target_contract.get("runner_labels")),
"request_packet_id": "gitea_runner_owner_attestation_request::ubuntu_latest",
"attestation_required": attestation_required,
"attestation_not_required": attestation_not_required,
"request_packet_id": (
"gitea_runner_owner_attestation_request::controlled_non110"
if attestation_not_required
else "gitea_runner_owner_attestation_request::ubuntu_latest"
),
"safe_next_step": (
"collect_redacted_owner_attestation_response_then_validate_no_runner_mutation"
"continue_public_gitea_queue_and_production_readback_without_runner_mutation"
if attestation_not_required
else "collect_redacted_owner_attestation_response_then_validate_no_runner_mutation"
),
},
"request_packet": {
"request_template_ready": request_ready,
"request_template_ready": request_packet_ready,
"attestation_required": attestation_required,
"attestation_not_required": attestation_not_required,
"request_sent": False,
"owner_response_received": False,
"owner_response_accepted": False,
"contract_id": _CONTRACT_ID,
"contract_id": str(target_contract.get("contract_id") or ""),
"display_name": str(target_contract.get("display_name") or ""),
"risk_level": str(target_contract.get("risk_level") or "high"),
"runner_labels": _strings(target_contract.get("runner_labels")),
@@ -127,8 +155,11 @@ def load_latest_gitea_workflow_runner_owner_attestation_request() -> dict[str, A
],
},
"rollups": {
"request_template_ready": request_ready,
"request_packet_count": int(request_ready),
"request_template_ready": request_packet_ready,
"attestation_required": attestation_required,
"attestation_not_required": attestation_not_required,
"request_packet_count": int(request_packet_ready),
"controlled_runner_contract_count": len(controlled_contracts),
"workflow_requiring_attestation_count": len(workflow_ids),
"runner_contracts_requiring_action_count": len(action_contracts),
"required_owner_field_count": len(_REQUIRED_FIELDS),
@@ -170,3 +201,12 @@ def _strings(value: Any) -> list[str]:
def _first_string(value: Any) -> str:
values = _strings(value)
return values[0] if values else ""
def _first_dict(value: Any) -> dict[str, Any]:
if not isinstance(value, list):
return {}
for item in value:
if isinstance(item, dict):
return item
return {}