99 lines
4.5 KiB
Python
99 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate a restricted Google mailbox acceptance record without reading its evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
SCENARIOS = {
|
|
"oauth-readonly",
|
|
"initial-import",
|
|
"duplicate-import",
|
|
"same-account-reconnect",
|
|
"different-account-rejected",
|
|
"provider-revocation",
|
|
"reviewed-send",
|
|
"gmail-threading",
|
|
"duplicate-approval",
|
|
"uncertain-send-reconciliation",
|
|
"sending-stop-control",
|
|
}
|
|
|
|
|
|
def require(condition: bool, message: str) -> None:
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def utc_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:
|
|
parsed = dt.datetime.fromisoformat(value.removesuffix("Z") + "+00:00")
|
|
except ValueError as error:
|
|
raise ValueError(f"{field} is not a valid timestamp.") from error
|
|
require(parsed.tzinfo == dt.timezone.utc, f"{field} must be UTC.")
|
|
return parsed
|
|
|
|
|
|
def validate(report: object) -> None:
|
|
require(isinstance(report, dict), "Acceptance record must be a JSON object.")
|
|
require(report.get("schemaVersion") == 1, "Unsupported acceptance record schema.")
|
|
require(report.get("system") == "google-mailbox", "Acceptance record system must be google-mailbox.")
|
|
require(re.fullmatch(r"[0-9a-f]{40}", str(report.get("releaseCommit", ""))) is not None,
|
|
"releaseCommit must be a full lowercase Git SHA.")
|
|
require(re.fullmatch(r"[0-9a-f]{64}", str(report.get("releaseRecordSha256", ""))) is not None,
|
|
"releaseRecordSha256 must be a SHA-256 digest.")
|
|
|
|
environment = str(report.get("environment", ""))
|
|
parsed_url = urlparse(environment)
|
|
require(parsed_url.scheme == "https" and parsed_url.hostname and parsed_url.path in ("", "/") and not parsed_url.query and not parsed_url.fragment,
|
|
"environment must be an HTTPS origin without credentials, path, query or fragment.")
|
|
require(parsed_url.username is None and parsed_url.password is None, "environment must not contain credentials.")
|
|
|
|
mailbox_label = str(report.get("mailboxLabel", ""))
|
|
require(3 <= len(mailbox_label) <= 80 and "@" not in mailbox_label,
|
|
"mailboxLabel must be a short non-email alias; do not put mailbox addresses in the record.")
|
|
require(2 <= len(str(report.get("operator", ""))) <= 120, "operator is required.")
|
|
started = utc_timestamp(report.get("startedAt"), "startedAt")
|
|
ended = utc_timestamp(report.get("endedAt"), "endedAt")
|
|
accepted = utc_timestamp(report.get("acceptedAt"), "acceptedAt")
|
|
require(started <= ended <= accepted, "Acceptance timestamps are out of order.")
|
|
require(2 <= len(str(report.get("acceptedBy", ""))) <= 120, "acceptedBy is required.")
|
|
|
|
scenarios = report.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)), "Scenario IDs must be unique objects.")
|
|
require(set(ids) == SCENARIOS, "Acceptance record does not contain the exact required scenario set.")
|
|
for item in scenarios:
|
|
scenario_id = item["id"]
|
|
require(item.get("status") == "pass", f"Scenario {scenario_id} has not passed.")
|
|
evidence = item.get("evidence")
|
|
require(isinstance(evidence, list) and 1 <= len(evidence) <= 10,
|
|
f"Scenario {scenario_id} requires one to ten restricted evidence references.")
|
|
require(all(isinstance(value, str) and 3 <= len(value) <= 200 and "@" not in value for value in evidence),
|
|
f"Scenario {scenario_id} has an invalid evidence reference; do not include email addresses or raw evidence.")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("record", type=Path)
|
|
args = parser.parse_args()
|
|
report = json.loads(args.record.read_text(encoding="utf-8"))
|
|
validate(report)
|
|
print(f"Google acceptance record is structurally complete: {len(SCENARIOS)} scenarios passed. This validates the record, not the underlying provider evidence.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (OSError, json.JSONDecodeError, ValueError) as error:
|
|
print(f"Google acceptance record rejected: {error}", file=__import__("sys").stderr)
|
|
raise SystemExit(1)
|