303 lines
13 KiB
Python
303 lines
13 KiB
Python
"""Repository-local invariants for the frozen Sense control-plane contract."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
import re
|
||
import unittest
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
OPENAPI_PATH = ROOT / "docs" / "contracts" / "sense-control-v1.openapi.json"
|
||
QUOTA_SQL_PATH = ROOT / "docs" / "contracts" / "site-quota-v1.sql"
|
||
CONTRACT_README_PATH = ROOT / "docs" / "contracts" / "README.md"
|
||
|
||
HTTP_METHODS = {"get", "post", "put", "patch", "delete", "options", "head", "trace"}
|
||
EXPECTED_OPERATIONS = {
|
||
("get", "/api/v1/sites/{site_id}/devices"),
|
||
("post", "/api/v1/sites/{site_id}/devices"),
|
||
("get", "/api/v1/sites/{site_id}/devices/{device_id}"),
|
||
("patch", "/api/v1/sites/{site_id}/devices/{device_id}"),
|
||
("put", "/api/v1/sites/{site_id}/devices/{device_id}/desired-state"),
|
||
("post", "/api/v1/sites/{site_id}/devices:batchDesiredState"),
|
||
("get", "/api/v1/operations/{operation_id}"),
|
||
}
|
||
|
||
EXPECTED_ENUMS = {
|
||
"Modality": ["video", "radar", "contact", "button", "wearable", "other"],
|
||
"Capability": ["video_capture", "audio_capture", "spatial_rule", "telemetry"],
|
||
"DesiredState": ["disabled", "enabled"],
|
||
"ActualState": ["pending", "online", "offline", "failed"],
|
||
}
|
||
|
||
REQUIRED_ERROR_CODES = {
|
||
"invalid_request",
|
||
"unauthenticated",
|
||
"forbidden",
|
||
"not_found",
|
||
"precondition_required",
|
||
"etag_mismatch",
|
||
"idempotency_conflict",
|
||
"duplicate_serial_number",
|
||
"quota_exceeded",
|
||
"quota_projection_unavailable",
|
||
"quota_projection_invalid",
|
||
"area_policy_denied",
|
||
"area_policy_unavailable",
|
||
"adapter_not_ready",
|
||
"endpoint_credentials_forbidden",
|
||
"batch_too_large",
|
||
}
|
||
|
||
|
||
def load_openapi() -> dict[str, Any]:
|
||
return json.loads(OPENAPI_PATH.read_text(encoding="utf-8"))
|
||
|
||
|
||
def resolve_local_ref(document: dict[str, Any], value: dict[str, Any]) -> dict[str, Any]:
|
||
"""Resolve one local JSON pointer; the contract only needs direct component refs."""
|
||
ref = value.get("$ref")
|
||
if not ref:
|
||
return value
|
||
if not ref.startswith("#/"):
|
||
raise ValueError(f"external ref is not allowed here: {ref}")
|
||
current: Any = document
|
||
for token in ref[2:].split("/"):
|
||
current = current[token.replace("~1", "/").replace("~0", "~")]
|
||
if not isinstance(current, dict):
|
||
raise ValueError(f"ref does not resolve to an object: {ref}")
|
||
return current
|
||
|
||
|
||
def operation_parameters(
|
||
document: dict[str, Any], operation: dict[str, Any]
|
||
) -> dict[str, dict[str, Any]]:
|
||
result: dict[str, dict[str, Any]] = {}
|
||
for value in operation.get("parameters", []):
|
||
parameter = resolve_local_ref(document, value)
|
||
result[parameter.get("name", "")] = parameter
|
||
return result
|
||
|
||
|
||
def validate_openapi(document: dict[str, Any]) -> list[str]:
|
||
errors: list[str] = []
|
||
if document.get("openapi") != "3.1.0":
|
||
errors.append("OpenAPI version must be 3.1.0")
|
||
if document.get("info", {}).get("version") != "1.0.0":
|
||
errors.append("contract version must be 1.0.0")
|
||
|
||
top_security = document.get("security")
|
||
if not top_security or not any("bearerAuth" in item for item in top_security):
|
||
errors.append("top-level bearerAuth is required")
|
||
|
||
operations: dict[tuple[str, str], dict[str, Any]] = {}
|
||
operation_ids: list[str] = []
|
||
for path, path_item in document.get("paths", {}).items():
|
||
if not path.startswith("/api/v1/"):
|
||
errors.append(f"unversioned path: {path}")
|
||
for method, operation in path_item.items():
|
||
if method not in HTTP_METHODS:
|
||
continue
|
||
key = (method, path)
|
||
operations[key] = operation
|
||
operation_id = operation.get("operationId")
|
||
if not operation_id:
|
||
errors.append(f"missing operationId: {method.upper()} {path}")
|
||
else:
|
||
operation_ids.append(operation_id)
|
||
if not operation.get("responses"):
|
||
errors.append(f"missing responses: {method.upper()} {path}")
|
||
effective_security = operation.get("security", top_security)
|
||
if not effective_security or not any(
|
||
"bearerAuth" in item for item in effective_security
|
||
):
|
||
errors.append(f"bearerAuth required: {method.upper()} {path}")
|
||
|
||
missing_operations = EXPECTED_OPERATIONS - set(operations)
|
||
unexpected_operations = set(operations) - EXPECTED_OPERATIONS
|
||
if missing_operations:
|
||
errors.append(f"missing operations: {sorted(missing_operations)}")
|
||
if unexpected_operations:
|
||
errors.append(f"unexpected operations: {sorted(unexpected_operations)}")
|
||
if len(operation_ids) != len(set(operation_ids)):
|
||
errors.append("operationId values must be unique")
|
||
|
||
list_operation = operations.get(("get", "/api/v1/sites/{site_id}/devices"), {})
|
||
list_parameters = operation_parameters(document, list_operation)
|
||
if "cursor" not in list_parameters:
|
||
errors.append("device list requires cursor pagination")
|
||
limit = list_parameters.get("limit", {}).get("schema", {})
|
||
if limit.get("default") != 50 or limit.get("maximum") != 100:
|
||
errors.append("device list limit must default to 50 and have maximum 100")
|
||
|
||
header_requirements = {
|
||
("post", "/api/v1/sites/{site_id}/devices"): "Idempotency-Key",
|
||
("post", "/api/v1/sites/{site_id}/devices:batchDesiredState"): "Idempotency-Key",
|
||
("patch", "/api/v1/sites/{site_id}/devices/{device_id}"): "If-Match",
|
||
("put", "/api/v1/sites/{site_id}/devices/{device_id}/desired-state"): "If-Match",
|
||
}
|
||
for key, header in header_requirements.items():
|
||
parameters = operation_parameters(document, operations.get(key, {}))
|
||
parameter = parameters.get(header, {})
|
||
if parameter.get("in") != "header" or parameter.get("required") is not True:
|
||
errors.append(f"{key[0].upper()} {key[1]} requires header {header}")
|
||
|
||
for key in (
|
||
("patch", "/api/v1/sites/{site_id}/devices/{device_id}"),
|
||
("put", "/api/v1/sites/{site_id}/devices/{device_id}/desired-state"),
|
||
):
|
||
responses = operations.get(key, {}).get("responses", {})
|
||
if "412" not in responses or "428" not in responses:
|
||
errors.append(f"{key[0].upper()} {key[1]} must define 412 and 428")
|
||
|
||
schemas = document.get("components", {}).get("schemas", {})
|
||
batch_items = (
|
||
schemas.get("BatchDesiredStateRequest", {})
|
||
.get("properties", {})
|
||
.get("items", {})
|
||
)
|
||
if batch_items.get("maxItems") != 128:
|
||
errors.append("batch desired-state request must have maximum 128 items")
|
||
|
||
client_schemas = ("DeviceCreate", "DevicePatch", "DesiredStateChange", "BatchDesiredStateRequest")
|
||
for schema_name in client_schemas:
|
||
if "tenant_id" in schemas.get(schema_name, {}).get("properties", {}):
|
||
errors.append(f"tenant_id must not be client-controlled in {schema_name}")
|
||
|
||
create_properties = schemas.get("DeviceCreate", {}).get("properties", {})
|
||
if "id" in create_properties:
|
||
errors.append("device id must be generated by the service")
|
||
required_create = set(schemas.get("DeviceCreate", {}).get("required", []))
|
||
if not {"modality", "capabilities"}.issubset(required_create):
|
||
errors.append("device create requires modality and capabilities")
|
||
|
||
for schema_name in ("DeviceCreate", "DevicePatch"):
|
||
properties = schemas.get(schema_name, {}).get("properties", {})
|
||
for field in ("endpoint_ref", "credential_ref", "profile_token"):
|
||
if properties.get(field, {}).get("writeOnly") is not True:
|
||
errors.append(f"{schema_name}.{field} must be writeOnly")
|
||
|
||
output_properties = schemas.get("Device", {}).get("properties", {})
|
||
tenant = output_properties.get("tenant_id", {})
|
||
if tenant.get("readOnly") is not True:
|
||
errors.append("Device.tenant_id must be readOnly")
|
||
forbidden_output_fields = {
|
||
"endpoint_ref",
|
||
"credential_ref",
|
||
"profile_token",
|
||
"active_profile",
|
||
"stream_uri",
|
||
"password",
|
||
"token",
|
||
}
|
||
leaked_fields = forbidden_output_fields & set(output_properties)
|
||
if leaked_fields:
|
||
errors.append(f"Device leaks sensitive fields: {sorted(leaked_fields)}")
|
||
|
||
for schema_name, expected in EXPECTED_ENUMS.items():
|
||
actual = schemas.get(schema_name, {}).get("enum")
|
||
if actual != expected:
|
||
errors.append(f"{schema_name} enum drift: {actual!r}")
|
||
|
||
error_codes = set(schemas.get("ErrorCode", {}).get("enum", []))
|
||
missing_error_codes = REQUIRED_ERROR_CODES - error_codes
|
||
if missing_error_codes:
|
||
errors.append(f"missing stable error codes: {sorted(missing_error_codes)}")
|
||
return errors
|
||
|
||
|
||
def validate_quota_sql(sql: str) -> list[str]:
|
||
errors: list[str] = []
|
||
normalized = re.sub(r"\s+", " ", sql.lower()).strip()
|
||
expected_signature = (
|
||
"create view bell.site_quota_v1 ( tenant_id, site_id, "
|
||
"max_video_channels, source_version, source_updated_at ) as"
|
||
)
|
||
if expected_signature not in normalized:
|
||
errors.append("bell.site_quota_v1 column signature is not frozen as expected")
|
||
if "alter view bell.site_quota_v1 owner to bell_app" not in normalized:
|
||
errors.append("Bell role must own the quota view")
|
||
if "grant usage on schema bell to sense_app" not in normalized:
|
||
errors.append("Sense role needs schema USAGE")
|
||
if "grant select on table bell.site_quota_v1 to sense_app" not in normalized:
|
||
errors.append("Sense role needs SELECT on only the quota view")
|
||
if "revoke all privileges on table bell.site_quota_v1 from public" not in normalized:
|
||
errors.append("PUBLIC privileges must be revoked")
|
||
if re.search(
|
||
r"grant\s+(?:all(?:\s+privileges)?|insert|update|delete|truncate|references|trigger)\b",
|
||
normalized,
|
||
):
|
||
errors.append("quota view grants a forbidden write or broad privilege")
|
||
if re.search(r"grant\s+select\s+on\s+table\s+bell\.sites\b", normalized):
|
||
errors.append("Sense must not receive SELECT on the Bell source table")
|
||
return errors
|
||
|
||
|
||
class SenseControlContractTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.document = load_openapi()
|
||
self.quota_sql = QUOTA_SQL_PATH.read_text(encoding="utf-8")
|
||
|
||
def test_frozen_openapi_invariants(self) -> None:
|
||
self.assertEqual([], validate_openapi(self.document))
|
||
|
||
def test_frozen_quota_projection_invariants(self) -> None:
|
||
self.assertEqual([], validate_quota_sql(self.quota_sql))
|
||
|
||
def test_contract_narrative_records_failure_and_compatibility_semantics(self) -> None:
|
||
text = CONTRACT_README_PATH.read_text(encoding="utf-8")
|
||
for marker in (
|
||
"默认 16",
|
||
"1～128",
|
||
"至少保存 24 小时",
|
||
"404 not_found",
|
||
"已有流保持运行",
|
||
"必须发布新版本",
|
||
):
|
||
self.assertIn(marker, text)
|
||
|
||
def test_validator_rejects_missing_authentication(self) -> None:
|
||
changed = copy.deepcopy(self.document)
|
||
changed.pop("security", None)
|
||
self.assertTrue(any("bearerAuth" in item for item in validate_openapi(changed)))
|
||
|
||
def test_validator_rejects_client_controlled_tenant(self) -> None:
|
||
changed = copy.deepcopy(self.document)
|
||
changed["components"]["schemas"]["DeviceCreate"]["properties"]["tenant_id"] = {
|
||
"type": "string"
|
||
}
|
||
self.assertTrue(any("tenant_id" in item for item in validate_openapi(changed)))
|
||
|
||
def test_validator_rejects_unbounded_list_or_batch(self) -> None:
|
||
changed = copy.deepcopy(self.document)
|
||
changed["components"]["parameters"]["Limit"]["schema"]["maximum"] = 101
|
||
changed["components"]["schemas"]["BatchDesiredStateRequest"]["properties"][
|
||
"items"
|
||
]["maxItems"] = 129
|
||
errors = validate_openapi(changed)
|
||
self.assertTrue(any("maximum 100" in item for item in errors))
|
||
self.assertTrue(any("maximum 128" in item for item in errors))
|
||
|
||
def test_validator_rejects_sensitive_output_or_writable_credential_ref(self) -> None:
|
||
changed = copy.deepcopy(self.document)
|
||
changed["components"]["schemas"]["DevicePatch"]["properties"]["credential_ref"][
|
||
"writeOnly"
|
||
] = False
|
||
changed["components"]["schemas"]["Device"]["properties"]["stream_uri"] = {
|
||
"type": "string"
|
||
}
|
||
errors = validate_openapi(changed)
|
||
self.assertTrue(any("credential_ref" in item for item in errors))
|
||
self.assertTrue(any("sensitive fields" in item for item in errors))
|
||
|
||
def test_validator_rejects_write_privilege(self) -> None:
|
||
changed = self.quota_sql + "\nGRANT ALL PRIVILEGES ON bell.site_quota_v1 TO sense_app;\n"
|
||
self.assertTrue(any("forbidden" in item for item in validate_quota_sql(changed)))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|