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

feat(api): expose gitea runner attestation request
Some checks failed
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 27s
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Your Name
2026-06-30 07:39:05 +08:00
parent 481a946d1b
commit 43396f5603
8 changed files with 419 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
apps/api/src/services/delivery_closure_workbench.py
View File

@@ -33,6 +33,9 @@ from src.services.credential_escrow_evidence_intake_readiness import (
from src.services.gitea_private_inventory_p0_scorecard import (
load_latest_gitea_private_inventory_p0_scorecard,
)
from src.services.gitea_workflow_runner_owner_attestation_request import (
load_latest_gitea_workflow_runner_owner_attestation_request,
)
from src.services.gitea_workflow_runner_health import (
load_latest_gitea_workflow_runner_health,
)
@@ -65,6 +68,9 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate()
)
gitea = load_latest_gitea_workflow_runner_health()
gitea_runner_attestation_request = (
load_latest_gitea_workflow_runner_owner_attestation_request()
)
runtime = load_latest_runtime_surface_inventory()
backup = load_latest_backup_dr_readiness_matrix()
credential_escrow_intake = load_latest_credential_escrow_evidence_intake_readiness()
@@ -78,6 +84,7 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
cicd_template_copy_receipt=cicd_template_copy_receipt,
cicd_runtime_enablement_gate=cicd_runtime_enablement_gate,
gitea=gitea,
gitea_runner_attestation_request=gitea_runner_attestation_request,
runtime=runtime,
backup=backup,
credential_escrow_intake=credential_escrow_intake,
@@ -95,6 +102,7 @@ def build_delivery_closure_workbench(
cicd_template_copy_receipt: dict[str, Any],
cicd_runtime_enablement_gate: dict[str, Any],
gitea: dict[str, Any],
gitea_runner_attestation_request: dict[str, Any],
runtime: dict[str, Any],
backup: dict[str, Any],
credential_escrow_intake: dict[str, Any],
@@ -145,6 +153,18 @@ def build_delivery_closure_workbench(
production_deploy_rollups = _dict(production_deploy.get("rollups"))
gitea_status = _dict(gitea.get("program_status"))
gitea_rollups = _dict(gitea.get("rollups"))
gitea_runner_request_readback = _dict(
gitea_runner_attestation_request.get("readback")
)
gitea_runner_request_packet = _dict(
gitea_runner_attestation_request.get("request_packet")
)
gitea_runner_request_rollups = _dict(
gitea_runner_attestation_request.get("rollups")
)
gitea_runner_request_boundaries = _dict(
gitea_runner_attestation_request.get("operation_boundaries")
)
runtime_status = _dict(runtime.get("program_status"))
runtime_rollups = _dict(runtime.get("rollups"))
backup_status = _dict(backup.get("program_status"))
@@ -1007,6 +1027,47 @@ def build_delivery_closure_workbench(
"metric": {
"kind": "workflow_count",
"count": _int(gitea_rollups.get("total_workflows")),
"runner_attestation_request_status": str(
gitea_runner_attestation_request.get("status") or ""
),
"runner_attestation_request_ready": (
gitea_runner_request_rollups.get("request_template_ready") is True
),
"runner_attestation_request_active_blocker_count": _int(
gitea_runner_request_rollups.get("active_blocker_count")
),
"runner_attestation_contract_id": str(
gitea_runner_request_readback.get("contract_id") or ""
),
"runner_attestation_workflow_count": _int(
gitea_runner_request_rollups.get(
"workflow_requiring_attestation_count"
)
),
"runner_attestation_required_owner_field_count": _int(
gitea_runner_request_rollups.get("required_owner_field_count")
),
"runner_attestation_owner_response_received_count": _int(
gitea_runner_request_rollups.get("owner_response_received_count")
),
"runner_attestation_owner_response_accepted_count": _int(
gitea_runner_request_rollups.get("owner_response_accepted_count")
),
"runner_attestation_request_send_performed": (
gitea_runner_request_boundaries.get("request_send_performed")
is True
),
"runner_attestation_runner_label_change_allowed": (
gitea_runner_request_boundaries.get("runner_label_change_allowed")
is True
),
"runner_attestation_runner_registration_allowed": (
gitea_runner_request_boundaries.get("runner_registration_allowed")
is True
),
"runner_attestation_secret_read_allowed": (
gitea_runner_request_boundaries.get("secret_read_allowed") is True
),
},
"href": "/deployments",
"next_action": _first_contract_action(gitea.get("runner_contracts")),
@@ -1363,6 +1424,55 @@ def build_delivery_closure_workbench(
)
is True
),
"gitea_runner_attestation_request_status": str(
gitea_runner_attestation_request.get("status") or ""
),
"gitea_runner_attestation_request_ready": (
gitea_runner_request_rollups.get("request_template_ready") is True
),
"gitea_runner_attestation_request_active_blocker_count": _int(
gitea_runner_request_rollups.get("active_blocker_count")
),
"gitea_runner_attestation_contract_id": str(
gitea_runner_request_readback.get("contract_id") or ""
),
"gitea_runner_attestation_runner_label": str(
gitea_runner_request_readback.get("runner_label") or ""
),
"gitea_runner_attestation_workflow_count": _int(
gitea_runner_request_rollups.get(
"workflow_requiring_attestation_count"
)
),
"gitea_runner_attestation_required_owner_field_count": _int(
gitea_runner_request_rollups.get("required_owner_field_count")
),
"gitea_runner_attestation_forbidden_action_count": _int(
gitea_runner_request_rollups.get("forbidden_action_count")
),
"gitea_runner_attestation_owner_response_received_count": _int(
gitea_runner_request_rollups.get("owner_response_received_count")
),
"gitea_runner_attestation_owner_response_accepted_count": _int(
gitea_runner_request_rollups.get("owner_response_accepted_count")
),
"gitea_runner_attestation_request_sent": (
gitea_runner_request_packet.get("request_sent") is True
),
"gitea_runner_attestation_request_send_performed": (
gitea_runner_request_boundaries.get("request_send_performed") is True
),
"gitea_runner_attestation_runner_label_change_allowed": (
gitea_runner_request_boundaries.get("runner_label_change_allowed")
is True
),
"gitea_runner_attestation_runner_registration_allowed": (
gitea_runner_request_boundaries.get("runner_registration_allowed")
is True
),
"gitea_runner_attestation_secret_read_allowed": (
gitea_runner_request_boundaries.get("secret_read_allowed") is True
),
"p0_cicd_baseline_status": str(cicd_baseline.get("status") or ""),
"p0_cicd_baseline_workplan_id": str(
cicd_baseline_readback.get("workplan_id") or ""

172
apps/api/src/services/gitea_workflow_runner_owner_attestation_request.py Normal file
View File

@@ -0,0 +1,172 @@
"""Gitea workflow runner owner attestation request readback.
Builds a metadata-only request packet from the committed Gitea workflow /
runner health contract. This module never sends the request, calls Gitea,
changes runner labels, registers runners, triggers workflows, or reads secrets.
"""
from __future__ import annotations
from typing import Any
from src.services.gitea_workflow_runner_health import (
load_latest_gitea_workflow_runner_health,
)
_SCHEMA_VERSION = "gitea_workflow_runner_owner_attestation_request_v1"
_CONTRACT_ID = "ubuntu_latest_gitea_runner_label"
_REQUIRED_FIELDS = [
"runner_label",
"actual_runner_mapping_ref",
"runner_host_alias",
"executor_scope",
"capacity_owner",
"maintenance_window",
"evidence_ref",
"no_secret_value_attestation",
"no_runner_registration_change_attestation",
"rollback_owner",
"post_check_owner",
]
_FORBIDDEN_ACTIONS = [
"workflow_modification",
"workflow_dispatch",
"runner_label_change",
"runner_registration",
"runner_restart",
"runner_container_stop",
"secret_or_runner_token_read",
"gitea_api_write",
"github_api",
]
def load_latest_gitea_workflow_runner_owner_attestation_request() -> dict[str, Any]:
"""Return the owner attestation request packet for runner label evidence."""
health = load_latest_gitea_workflow_runner_health()
rollups = _dict(health.get("rollups"))
action_contracts = [
contract
for contract in health.get("runner_contracts", [])
if isinstance(contract, dict)
and contract.get("contract_id")
in set(_strings(rollups.get("runner_contracts_requiring_action")))
]
target_contract = next(
(
contract
for contract in action_contracts
if contract.get("contract_id") == _CONTRACT_ID
),
{},
)
workflow_ids = _strings(rollups.get("workflow_ids_requiring_runner_attestation"))
request_ready = bool(target_contract) and bool(workflow_ids)
active_blockers = []
if not request_ready:
active_blockers.append("runner_attestation_request_source_missing")
active_blockers.append("owner_attestation_response_not_received")
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"
),
"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,
"runner_label": _first_string(target_contract.get("runner_labels")),
"request_packet_id": "gitea_runner_owner_attestation_request::ubuntu_latest",
"safe_next_step": (
"collect_redacted_owner_attestation_response_then_validate_no_runner_mutation"
),
},
"request_packet": {
"request_template_ready": request_ready,
"request_sent": False,
"owner_response_received": False,
"owner_response_accepted": False,
"contract_id": _CONTRACT_ID,
"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")),
"workflow_ids_requiring_attestation": workflow_ids,
"required_owner_fields": _REQUIRED_FIELDS,
"forbidden_actions": _FORBIDDEN_ACTIONS,
"allowed_response_shape": {
"redacted_metadata_only": True,
"evidence_ref_required": True,
"secret_value_allowed": False,
"runner_token_allowed": False,
"authorization_header_allowed": False,
"raw_runner_config_allowed": False,
},
"acceptance_checks": [
"runner_label_mapping_ref_present",
"runner_host_alias_present",
"capacity_owner_present",
"maintenance_window_present",
"no_secret_value_attested",
"no_runner_registration_change_attested",
"post_check_owner_present",
],
"rejection_rules": [
"contains_secret_value",
"contains_runner_token",
"requests_runner_label_change",
"requests_runner_registration",
"requests_workflow_dispatch",
"missing_evidence_ref",
],
},
"rollups": {
"request_template_ready": request_ready,
"request_packet_count": int(request_ready),
"workflow_requiring_attestation_count": len(workflow_ids),
"runner_contracts_requiring_action_count": len(action_contracts),
"required_owner_field_count": len(_REQUIRED_FIELDS),
"forbidden_action_count": len(_FORBIDDEN_ACTIONS),
"owner_response_received_count": 0,
"owner_response_accepted_count": 0,
"active_blocker_count": len(active_blockers),
"runtime_gate_opened": False,
},
"active_blockers": active_blockers,
"operation_boundaries": {
"read_only_api_allowed": True,
"request_send_performed": False,
"workflow_modification_allowed": False,
"workflow_trigger_allowed": False,
"runner_label_change_allowed": False,
"runner_registration_allowed": False,
"runner_restart_allowed": False,
"runner_container_stop_allowed": False,
"secret_read_allowed": False,
"secret_plaintext_allowed": False,
"runner_token_read_allowed": False,
"gitea_api_write_allowed": False,
"github_api_used": False,
},
}
def _dict(value: Any) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _strings(value: Any) -> list[str]:
if not isinstance(value, list):
return []
return [str(item) for item in value if item is not None]
def _first_string(value: Any) -> str:
values = _strings(value)
return values[0] if values else ""