117 lines
5.9 KiB
Python
117 lines
5.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate restricted Gate B identity, preference and privacy acceptance evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
SCENARIOS = {
|
|
"role-boundaries", "account-lifecycle", "login-throttling", "session-invalidation",
|
|
"preference-coverage", "data-inventory", "retention-deletion", "backup-retention",
|
|
"provider-processing", "audit-review", "known-identity-limitations",
|
|
}
|
|
PREFERENCES = {
|
|
"hotel-name", "timezone", "signature", "ai-drafts", "staff-sending",
|
|
"faq-mode", "pms-updates", "payment-creation",
|
|
}
|
|
|
|
|
|
def require(condition: bool, message: str) -> None:
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def timestamp(value: object, field: str) -> dt.datetime:
|
|
require(isinstance(value, str) and value.endswith("Z"), f"{field} must be a UTC timestamp ending in Z.")
|
|
try:
|
|
return dt.datetime.fromisoformat(value.removesuffix("Z") + "+00:00")
|
|
except ValueError as error:
|
|
raise ValueError(f"{field} is not a valid timestamp.") from error
|
|
|
|
|
|
def name(value: object, field: str) -> str:
|
|
result = str(value or "").strip()
|
|
require(2 <= len(result) <= 120 and "@" not in result, f"{field} requires a name without an email address.")
|
|
return result
|
|
|
|
|
|
def refs(value: object, field: str) -> None:
|
|
require(isinstance(value, list) and 1 <= len(value) <= 10 and all(
|
|
isinstance(item, str) and 3 <= len(item) <= 200 and "@" not in item for item in value
|
|
), f"{field} requires safe opaque evidence references.")
|
|
|
|
|
|
def validate(record: object) -> None:
|
|
require(isinstance(record, dict), "Acceptance record must be a JSON object.")
|
|
require(record.get("schemaVersion") == 1, "Unsupported identity/privacy acceptance schema.")
|
|
require(record.get("system") == "guestops-identity-privacy", "system must be guestops-identity-privacy.")
|
|
require(record.get("targetGate") == "B", "Identity/privacy acceptance must target Gate B.")
|
|
require(re.fullmatch(r"[0-9a-f]{40}", str(record.get("releaseCommit", ""))) is not None, "releaseCommit must be a full lowercase Git SHA.")
|
|
require(re.fullmatch(r"[0-9a-f]{64}", str(record.get("releaseRecordSha256", ""))) is not None, "releaseRecordSha256 must be a SHA-256 digest.")
|
|
origin = urlparse(str(record.get("environment", "")))
|
|
require(origin.scheme == "https" and origin.hostname and origin.path in ("", "/") and not origin.query and not origin.fragment and origin.username is None and origin.password is None,
|
|
"environment must be an HTTPS origin without credentials, path, query or fragment.")
|
|
operator = name(record.get("operator"), "operator")
|
|
reviewer = name(record.get("reviewedBy"), "reviewedBy")
|
|
require(operator.casefold() != reviewer.casefold(), "operator and reviewedBy must be different people.")
|
|
started = timestamp(record.get("startedAt"), "startedAt")
|
|
ended = timestamp(record.get("endedAt"), "endedAt")
|
|
reviewed = timestamp(record.get("reviewedAt"), "reviewedAt")
|
|
require(started <= ended <= reviewed, "Acceptance timestamps are out of order.")
|
|
|
|
retention = record.get("retention")
|
|
require(isinstance(retention, dict), "retention decisions are required.")
|
|
for field in ("conversationDays", "auditDays", "backupDays", "accountDays"):
|
|
value = retention.get(field)
|
|
require(isinstance(value, int) and not isinstance(value, bool) and 1 <= value <= 3650,
|
|
f"retention.{field} must be between 1 and 3650 days.")
|
|
for field in ("privacyOwner", "deletionOwner", "legalHoldOwner"):
|
|
name(retention.get(field), f"retention.{field}")
|
|
for field in ("deletionProcedure", "legalHoldProcedure"):
|
|
refs([retention.get(field)], f"retention.{field}")
|
|
|
|
providers = record.get("providers")
|
|
require(isinstance(providers, dict) and set(providers) == {"google", "openai"},
|
|
"providers must contain exactly google and openai decisions.")
|
|
require(all(isinstance(decision, dict) for decision in providers.values()),
|
|
"Each provider decision must be an object.")
|
|
require(providers["google"].get("status") == "accepted", "Google processing must be accepted for the Gate B mailbox pilot.")
|
|
require(providers["openai"].get("status") in ("accepted", "disabled"), "OpenAI processing must be accepted or disabled.")
|
|
for provider, decision in providers.items():
|
|
refs(decision.get("evidence"), f"Provider {provider}")
|
|
preferences = record.get("preferencesReviewed")
|
|
require(isinstance(preferences, list) and all(isinstance(item, str) for item in preferences)
|
|
and set(preferences) == PREFERENCES and len(preferences) == len(PREFERENCES),
|
|
"preferencesReviewed must contain the exact owner-controlled preference set.")
|
|
|
|
scenarios = record.get("scenarios")
|
|
require(isinstance(scenarios, list), "scenarios must be a list.")
|
|
ids = [item.get("id") for item in scenarios if isinstance(item, dict)]
|
|
require(len(ids) == len(scenarios) and len(ids) == len(set(ids)) and set(ids) == SCENARIOS,
|
|
"Acceptance record requires the exact identity/privacy scenario set.")
|
|
for item in scenarios:
|
|
require(item.get("status") == "pass", f"Scenario {item['id']} has not passed.")
|
|
refs(item.get("evidence"), f"Scenario {item['id']}")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("record", type=Path)
|
|
args = parser.parse_args()
|
|
validate(json.loads(args.record.read_text(encoding="utf-8")))
|
|
print("Identity/privacy acceptance record is structurally complete. This validates the record, not its restricted evidence or legal decisions.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (OSError, ValueError, json.JSONDecodeError) as error:
|
|
print(f"Identity/privacy acceptance record rejected: {error}", file=__import__("sys").stderr)
|
|
raise SystemExit(1)
|