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

fix(security): verify workload DB concurrency budget
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 2m19s
Details
CD Pipeline / build-and-deploy (push) Successful in 6m54s
Details
CD Pipeline / post-deploy-checks (push) Successful in 2m6s
Details
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 1s
Details
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 48s
Details

Browse Source
This commit is contained in:
ogt
2026-07-11 04:44:23 +08:00
parent 2d786da147
commit 0660813781
6 changed files with 206 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
.gitea/workflows/cd.yaml
View File

@@ -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" \

30
apps/api/src/services/executor_trust_boundary_readback.py
View File

@@ -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": (

50
apps/api/tests/test_executor_trust_boundary_readback.py
View File

@@ -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"]
)

10
apps/api/tests/test_signal_worker_ansible_executor_binding.py
View File

@@ -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

6
apps/api/tests/test_workload_database_identity_projection.py
View File

@@ -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

37
scripts/ops/awoooi-workload-db-identity-bootstrap.sh
View File

@@ -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