diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index 58d9ed40c..ae1ee238b 100644 --- a/.gitea/workflows/cd.yaml +++ b/.gitea/workflows/cd.yaml @@ -2801,11 +2801,15 @@ jobs: read_workload_db_identity() { workload="$1" container="$2" + concurrent_probe_count="$3" $KUBECTL exec -i "deployment/$workload" -n awoooi-prod \ - -c "$container" -- python - <<'PY' + -c "$container" -- env \ + AWOOOI_DB_CONCURRENT_PROBE_COUNT="$concurrent_probe_count" \ + python - <<'PY' import asyncio import hashlib import json + import os from sqlalchemy import text @@ -2865,10 +2869,24 @@ jobs: AND NOT has_sequence_privilege(current_user, sequence.oid, 'UPDATE')) ) """))).scalar_one()) + + concurrent_probe_count = int( + os.environ["AWOOOI_DB_CONCURRENT_PROBE_COUNT"] + ) + + async def probe_connection() -> None: + async with get_db_context("awoooi") as probe_db: + await probe_db.execute(text("SELECT pg_sleep(0.25)")) + + await asyncio.gather(*( + probe_connection() + for _ in range(concurrent_probe_count) + )) except Exception as exc: print(json.dumps({ "role_fingerprint": "", "connection_limit": None, + "concurrent_connection_probe_count": 0, "representative_preflight_passed": False, "table_privilege_gap_count": None, "sequence_privilege_gap_count": None, @@ -2880,6 +2898,7 @@ jobs: str(row["role_name"]).encode("utf-8") ).hexdigest(), "connection_limit": int(row["connection_limit"]), + "concurrent_connection_probe_count": concurrent_probe_count, "representative_preflight_passed": ( table_gap_count == 0 and sequence_gap_count == 0 ), @@ -2920,12 +2939,12 @@ jobs: fi } - API_DB_IDENTITY=$(read_workload_db_identity awoooi-api api \ + API_DB_IDENTITY=$(read_workload_db_identity awoooi-api api 8 \ | tail -n 1 || true) - WORKER_DB_IDENTITY=$(read_workload_db_identity awoooi-worker worker \ + WORKER_DB_IDENTITY=$(read_workload_db_identity awoooi-worker worker 5 \ | tail -n 1 || true) BROKER_DB_IDENTITY=$(read_workload_db_identity \ - awoooi-ansible-executor-broker broker | tail -n 1 || true) + awoooi-ansible-executor-broker broker 2 | tail -n 1 || true) API_DB_ROLE_FINGERPRINT=$(printf '%s' "$API_DB_IDENTITY" \ | json_field role_fingerprint 2>/dev/null || true) WORKER_DB_ROLE_FINGERPRINT=$(printf '%s' "$WORKER_DB_IDENTITY" \ @@ -2938,6 +2957,12 @@ jobs: | json_field connection_limit 2>/dev/null || true) BROKER_DB_CONNECTION_LIMIT=$(printf '%s' "$BROKER_DB_IDENTITY" \ | json_field connection_limit 2>/dev/null || true) + API_DB_CONCURRENT_PROBE_COUNT=$(printf '%s' "$API_DB_IDENTITY" \ + | json_field concurrent_connection_probe_count 2>/dev/null || true) + WORKER_DB_CONCURRENT_PROBE_COUNT=$(printf '%s' "$WORKER_DB_IDENTITY" \ + | json_field concurrent_connection_probe_count 2>/dev/null || true) + BROKER_DB_CONCURRENT_PROBE_COUNT=$(printf '%s' "$BROKER_DB_IDENTITY" \ + | json_field concurrent_connection_probe_count 2>/dev/null || true) API_DB_REPRESENTATIVE_PREFLIGHT=$(normalize_bool "$(printf '%s' "$API_DB_IDENTITY" \ | json_field representative_preflight_passed 2>/dev/null || true)") WORKER_DB_REPRESENTATIVE_PREFLIGHT=$(normalize_bool "$(printf '%s' "$WORKER_DB_IDENTITY" \ @@ -2959,6 +2984,9 @@ jobs: API_DB_CONNECTION_LIMIT=${API_DB_CONNECTION_LIMIT:-0} WORKER_DB_CONNECTION_LIMIT=${WORKER_DB_CONNECTION_LIMIT:-0} BROKER_DB_CONNECTION_LIMIT=${BROKER_DB_CONNECTION_LIMIT:-0} + API_DB_CONCURRENT_PROBE_COUNT=${API_DB_CONCURRENT_PROBE_COUNT:-0} + WORKER_DB_CONCURRENT_PROBE_COUNT=${WORKER_DB_CONCURRENT_PROBE_COUNT:-0} + BROKER_DB_CONCURRENT_PROBE_COUNT=${BROKER_DB_CONCURRENT_PROBE_COUNT:-0} API_DB_TABLE_PRIVILEGE_GAP_COUNT=${API_DB_TABLE_PRIVILEGE_GAP_COUNT:-1} WORKER_DB_TABLE_PRIVILEGE_GAP_COUNT=${WORKER_DB_TABLE_PRIVILEGE_GAP_COUNT:-1} BROKER_DB_TABLE_PRIVILEGE_GAP_COUNT=${BROKER_DB_TABLE_PRIVILEGE_GAP_COUNT:-1} @@ -3025,6 +3053,9 @@ jobs: if [ "$API_DB_REPRESENTATIVE_PREFLIGHT" = "true" ] \ && [ "$WORKER_DB_REPRESENTATIVE_PREFLIGHT" = "true" ] \ && [ "$BROKER_DB_REPRESENTATIVE_PREFLIGHT" = "true" ] \ + && [ "$API_DB_CONCURRENT_PROBE_COUNT" -ge 8 ] \ + && [ "$WORKER_DB_CONCURRENT_PROBE_COUNT" -ge 5 ] \ + && [ "$BROKER_DB_CONCURRENT_PROBE_COUNT" -ge 2 ] \ && [ "$API_DB_TABLE_PRIVILEGE_GAP_COUNT" -eq 0 ] \ && [ "$WORKER_DB_TABLE_PRIVILEGE_GAP_COUNT" -eq 0 ] \ && [ "$BROKER_DB_TABLE_PRIVILEGE_GAP_COUNT" -eq 0 ] \ @@ -3039,12 +3070,54 @@ jobs: && [ "$API_DB_NULL_POOL" = "true" ] \ && [ "$WORKER_DB_NULL_POOL" = "true" ] \ && [ "$BROKER_DB_NULL_POOL" = "true" ] \ - && [ "$API_DB_CONNECTION_LIMIT" -ge 4 ] \ - && [ "$WORKER_DB_CONNECTION_LIMIT" -ge 2 ] \ - && [ "$BROKER_DB_CONNECTION_LIMIT" -ge 2 ]; then + && [ "$API_DB_CONNECTION_LIMIT" -ge 12 ] \ + && [ "$WORKER_DB_CONNECTION_LIMIT" -ge 8 ] \ + && [ "$BROKER_DB_CONNECTION_LIMIT" -ge 4 ]; then CONNECTION_BUDGET_VERIFIED=true fi + API_HTTP_CONCURRENCY_VERIFIED=$(python3 - <<'PY' + import concurrent.futures + import time + import urllib.request + + urls = ( + "https://awoooi.wooo.work/api/v1/platform/events/dossier/coverage?project_id=awoooi", + "https://awoooi.wooo.work/api/v1/platform/cicd/events?project_id=awoooi", + ) * 6 + + + def fetch(url: str) -> int: + with urllib.request.urlopen(url, timeout=20) as response: + return int(response.status) + + + last_result = "unavailable" + for attempt in range(1, 4): + try: + with concurrent.futures.ThreadPoolExecutor( + max_workers=8 + ) as executor: + statuses = list(executor.map(fetch, urls)) + except Exception as exc: + last_result = type(exc).__name__ + else: + last_result = ",".join(str(status) for status in statuses) + if all(status == 200 for status in statuses): + print("true") + break + if attempt < 3: + time.sleep(5) + else: + raise SystemExit( + "API database concurrency probe failed: " + last_result + ) + PY + ) + if [ "$API_HTTP_CONCURRENCY_VERIFIED" != "true" ]; then + CONNECTION_BUDGET_VERIFIED=false + fi + BOUNDARY_VERIFIED_AT=$(date -u +"%Y-%m-%dT%H:%M:%SZ") $KUBECTL create configmap awoooi-executor-boundary-verification \ -n awoooi-prod \ @@ -3077,9 +3150,12 @@ jobs: --from-literal=api_db_connection_limit="$API_DB_CONNECTION_LIMIT" \ --from-literal=worker_db_connection_limit="$WORKER_DB_CONNECTION_LIMIT" \ --from-literal=broker_db_connection_limit="$BROKER_DB_CONNECTION_LIMIT" \ - --from-literal=api_required_connection_budget=4 \ - --from-literal=worker_required_connection_budget=2 \ + --from-literal=api_required_connection_budget=8 \ + --from-literal=worker_required_connection_budget=5 \ --from-literal=broker_required_connection_budget=2 \ + --from-literal=api_concurrent_connection_probe_count="$API_DB_CONCURRENT_PROBE_COUNT" \ + --from-literal=worker_concurrent_connection_probe_count="$WORKER_DB_CONCURRENT_PROBE_COUNT" \ + --from-literal=broker_concurrent_connection_probe_count="$BROKER_DB_CONCURRENT_PROBE_COUNT" \ --from-literal=api_representative_db_preflight_passed="$API_DB_REPRESENTATIVE_PREFLIGHT" \ --from-literal=worker_representative_db_preflight_passed="$WORKER_DB_REPRESENTATIVE_PREFLIGHT" \ --from-literal=broker_representative_db_preflight_passed="$BROKER_DB_REPRESENTATIVE_PREFLIGHT" \ @@ -3090,6 +3166,7 @@ jobs: --from-literal=worker_sequence_privilege_gap_count="$WORKER_DB_SEQUENCE_PRIVILEGE_GAP_COUNT" \ --from-literal=broker_sequence_privilege_gap_count="$BROKER_DB_SEQUENCE_PRIVILEGE_GAP_COUNT" \ --from-literal=representative_db_preflights_verified="$REPRESENTATIVE_DB_PREFLIGHTS_VERIFIED" \ + --from-literal=api_http_concurrency_probe_passed="$API_HTTP_CONCURRENCY_VERIFIED" \ --from-literal=api_database_null_pool="$API_DB_NULL_POOL" \ --from-literal=worker_database_null_pool="$WORKER_DB_NULL_POOL" \ --from-literal=broker_database_null_pool="$BROKER_DB_NULL_POOL" \ diff --git a/apps/api/src/services/executor_trust_boundary_readback.py b/apps/api/src/services/executor_trust_boundary_readback.py index 5742e32a6..e75daa202 100644 --- a/apps/api/src/services/executor_trust_boundary_readback.py +++ b/apps/api/src/services/executor_trust_boundary_readback.py @@ -74,6 +74,11 @@ def _build_database_identity_boundary( "required_connection_budget": _int_or_none( receipt.get(f"{workload_id}_required_connection_budget") ), + "concurrent_connection_probe_count": _int_or_none( + receipt.get( + f"{workload_id}_concurrent_connection_probe_count" + ) + ), "database_null_pool": _truthy( receipt.get(f"{workload_id}_database_null_pool") ), @@ -130,13 +135,27 @@ def _build_database_identity_boundary( and workloads[workload_id]["required_connection_budget"] > 0 for workload_id in _DB_WORKLOAD_IDS ) + api_http_concurrency_verified = _truthy( + receipt.get("api_http_concurrency_probe_passed") + ) representative_preflights_verified = bool( all( workloads[workload_id]["representative_preflight_passed"] is True + and isinstance( + workloads[workload_id]["required_connection_budget"], int + ) + and isinstance( + workloads[workload_id]["concurrent_connection_probe_count"], + int, + ) + and workloads[workload_id]["required_connection_budget"] > 0 + and workloads[workload_id]["concurrent_connection_probe_count"] + >= workloads[workload_id]["required_connection_budget"] and workloads[workload_id]["table_privilege_gap_count"] == 0 and workloads[workload_id]["sequence_privilege_gap_count"] == 0 for workload_id in _DB_WORKLOAD_IDS ) + and api_http_concurrency_verified and _truthy(receipt.get("representative_db_preflights_verified")) ) db_identity_isolated = bool( @@ -162,6 +181,8 @@ def _build_database_identity_boundary( blockers.append("database_secret_refs_not_distinct") if not distinct_role_fingerprints: blockers.append("database_role_fingerprints_not_distinct") + if not api_http_concurrency_verified: + blockers.append("api_http_concurrency_probe_not_verified") for workload_id in _DB_WORKLOAD_IDS: workload = workloads[workload_id] if workload["role_fingerprint"] is None: @@ -170,6 +191,12 @@ def _build_database_identity_boundary( blockers.append(f"{workload_id}_database_null_pool_not_verified") if ( workload["representative_preflight_passed"] is not True + or not isinstance( + workload["concurrent_connection_probe_count"], int + ) + or not isinstance(workload["required_connection_budget"], int) + or workload["concurrent_connection_probe_count"] + < workload["required_connection_budget"] or workload["table_privilege_gap_count"] != 0 or workload["sequence_privilege_gap_count"] != 0 ): @@ -220,6 +247,9 @@ def _build_database_identity_boundary( "representative_db_preflights_verified": ( representative_preflights_verified ), + "api_http_concurrency_probe_passed": ( + api_http_concurrency_verified + ), }, "connection_budget": { "observed_total": ( diff --git a/apps/api/tests/test_executor_trust_boundary_readback.py b/apps/api/tests/test_executor_trust_boundary_readback.py index 43e3ca6a7..82bdac559 100644 --- a/apps/api/tests/test_executor_trust_boundary_readback.py +++ b/apps/api/tests/test_executor_trust_boundary_readback.py @@ -35,12 +35,15 @@ def _verified_receipt(source_sha: str = "abc123") -> dict[str, str]: "api_db_role_fingerprint": "a" * 64, "worker_db_role_fingerprint": "b" * 64, "broker_db_role_fingerprint": "c" * 64, - "api_db_connection_limit": "4", - "worker_db_connection_limit": "2", - "broker_db_connection_limit": "2", - "api_required_connection_budget": "4", - "worker_required_connection_budget": "2", + "api_db_connection_limit": "12", + "worker_db_connection_limit": "8", + "broker_db_connection_limit": "4", + "api_required_connection_budget": "8", + "worker_required_connection_budget": "5", "broker_required_connection_budget": "2", + "api_concurrent_connection_probe_count": "8", + "worker_concurrent_connection_probe_count": "5", + "broker_concurrent_connection_probe_count": "2", "api_representative_db_preflight_passed": "true", "worker_representative_db_preflight_passed": "true", "broker_representative_db_preflight_passed": "true", @@ -51,6 +54,7 @@ def _verified_receipt(source_sha: str = "abc123") -> dict[str, str]: "worker_sequence_privilege_gap_count": "0", "broker_sequence_privilege_gap_count": "0", "representative_db_preflights_verified": "true", + "api_http_concurrency_probe_passed": "true", "api_database_null_pool": "true", "worker_database_null_pool": "true", "broker_database_null_pool": "true", @@ -80,8 +84,8 @@ def test_executor_trust_boundary_requires_live_controls_and_matching_source() -> assert database_boundary["db_identity_isolated"] is True assert database_boundary["connection_budget_verified"] is True assert database_boundary["connection_budget"] == { - "observed_total": 8, - "required_total": 8, + "observed_total": 24, + "required_total": 15, } assert database_boundary["controls"][ "representative_db_preflights_verified" @@ -161,3 +165,35 @@ def test_executor_trust_boundary_requires_representative_database_preflight() -> assert "worker_representative_db_preflight_not_verified" in ( database_boundary["active_blockers"] ) + + +def test_executor_trust_boundary_rejects_static_budget_without_concurrency_probe() -> None: + receipt = _verified_receipt() + receipt["api_concurrent_connection_probe_count"] = "4" + + readback = build_executor_trust_boundary_readback( + receipt, + deployed_source_sha="abc123", + ) + + database_boundary = readback["database_identity_boundary"] + assert database_boundary["connection_budget_verified"] is False + assert "api_representative_db_preflight_not_verified" in ( + database_boundary["active_blockers"] + ) + + +def test_executor_trust_boundary_requires_api_http_concurrency_probe() -> None: + receipt = _verified_receipt() + receipt["api_http_concurrency_probe_passed"] = "false" + + readback = build_executor_trust_boundary_readback( + receipt, + deployed_source_sha="abc123", + ) + + database_boundary = readback["database_identity_boundary"] + assert database_boundary["connection_budget_verified"] is False + assert "api_http_concurrency_probe_not_verified" in ( + database_boundary["active_blockers"] + ) diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py index e5db47418..171539536 100644 --- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -173,9 +173,15 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "current_user AS role_name" in workflow assert "hashlib.sha256" in workflow assert "representative_db_preflights_verified" in workflow - assert "api_required_connection_budget=4" in workflow - assert "worker_required_connection_budget=2" in workflow + assert "api_required_connection_budget=8" in workflow + assert "worker_required_connection_budget=5" in workflow assert "broker_required_connection_budget=2" in workflow + assert "api_concurrent_connection_probe_count" in workflow + assert "worker_concurrent_connection_probe_count" in workflow + assert "broker_concurrent_connection_probe_count" in workflow + assert "api_http_concurrency_probe_passed" in workflow + assert "ThreadPoolExecutor" in workflow + assert "max_workers=8" in workflow assert "WORKLOAD_DB_ROLLBACK_FILE" in workflow assert "restore_workload_database_projection" in workflow assert "awoooi-workload-db-identity-bootstrap.sh" in workflow diff --git a/apps/api/tests/test_workload_database_identity_projection.py b/apps/api/tests/test_workload_database_identity_projection.py index 702b03cc9..0962a796e 100644 --- a/apps/api/tests/test_workload_database_identity_projection.py +++ b/apps/api/tests/test_workload_database_identity_projection.py @@ -102,6 +102,12 @@ def test_workload_database_bootstrap_keeps_secret_values_off_output() -> None: assert "docker exec -i -u postgres" in script assert "table_privilege_gap_count" in script assert "sequence_privilege_gap_count" in script + assert "REQUIRED_CONCURRENT_CONNECTIONS" in script + assert "concurrent_connection_probe_count" in script + assert 'text("SELECT pg_sleep(0.25)")' in script + assert "api) printf '12'" in script + assert "worker) printf '8'" in script + assert "broker) printf '4'" in script assert '"incidents"' in script assert '"knowledge_entries"' in script assert '"awooop_run_state"' in script diff --git a/scripts/ops/awoooi-workload-db-identity-bootstrap.sh b/scripts/ops/awoooi-workload-db-identity-bootstrap.sh index b2920efa3..189187d1a 100755 --- a/scripts/ops/awoooi-workload-db-identity-bootstrap.sh +++ b/scripts/ops/awoooi-workload-db-identity-bootstrap.sh @@ -61,8 +61,18 @@ workload_secret() { workload_connection_limit() { case "$1" in - api) printf '4' ;; - worker|broker) printf '2' ;; + api) printf '12' ;; + worker) printf '8' ;; + broker) printf '4' ;; + *) return 1 ;; + esac +} + +workload_required_connection_budget() { + case "$1" in + api) printf '8' ;; + worker) printf '5' ;; + broker) printf '2' ;; *) return 1 ;; esac } @@ -211,6 +221,7 @@ preflight_workload() { workload="$1" secret="$2" connection_limit="$3" + required_connection_budget="$4" deployment_spec="$(workload_deployment "$workload")" deployment="${deployment_spec%%|*}" container="${deployment_spec#*|}" @@ -251,6 +262,8 @@ spec: key: DATABASE_URL - name: EXPECTED_CONNECTION_LIMIT value: "${connection_limit}" + - name: REQUIRED_CONCURRENT_CONNECTIONS + value: "${required_connection_budget}" - name: WORKLOAD_ID value: "${workload}" command: ["python", "-c"] @@ -320,6 +333,19 @@ spec: AND NOT has_sequence_privilege(current_user, sequence.oid, 'UPDATE')) ) """))).scalar_one()) + + required_concurrency = int( + os.environ["REQUIRED_CONCURRENT_CONNECTIONS"] + ) + + async def probe_connection() -> None: + async with engine.connect() as probe: + await probe.execute(text("SELECT pg_sleep(0.25)")) + + await asyncio.gather(*( + probe_connection() + for _ in range(required_concurrency) + )) finally: await engine.dispose() @@ -336,6 +362,7 @@ spec: str(identity["role_name"]).encode("utf-8") ).hexdigest(), "connection_limit": observed_limit, + "concurrent_connection_probe_count": required_concurrency, "representative_read_count": 3, "table_privilege_gap_count": table_gap_count, "sequence_privilege_gap_count": sequence_gap_count, @@ -410,6 +437,7 @@ for workload in "${selected_workloads[@]}"; do role="$(workload_role "$workload")" secret="$(workload_secret "$workload")" connection_limit="$(workload_connection_limit "$workload")" + required_connection_budget="$(workload_required_connection_budget "$workload")" secret_present=false if kube get secret "$secret" -n "$NAMESPACE" >/dev/null 2>&1; then secret_present=true @@ -443,10 +471,11 @@ for workload in "${selected_workloads[@]}"; do else ready=false fi - echo "workload=$workload secret_ref=$secret secret_present=$secret_present role_present=$role_present contract_ready=$state_ready connection_limit=$connection_limit" + echo "workload=$workload secret_ref=$secret secret_present=$secret_present role_present=$role_present contract_ready=$state_ready connection_limit=$connection_limit required_connection_budget=$required_connection_budget" if [ "$MODE" = "apply" ] || [ "$MODE" = "verify" ]; then - preflight_workload "$workload" "$secret" "$connection_limit" + preflight_workload "$workload" "$secret" "$connection_limit" \ + "$required_connection_budget" fi done