#!/usr/bin/env python3 """privacy_check.py - BL-157 (2026-09-26; Phase B 2026-09-27). Read-only. The privacy page (Hosting/site/privacy.html, and CompMaps' own) is a claim about settings. This checks the settings, so a later change made for a good reason (backups, access logs, a debugging print) cannot quietly make the page false. Run it in the deploy routine; exit 1 on any finding. py -3.14 Hosting/privacy_check.py Needs the AWS CLI with the operator's credentials (it only reads). """ import json import pathlib import re import subprocess import sys import tempfile R = "ca-central-1" HERE = pathlib.Path(__file__).resolve().parent TABLE = "site-maps-jobs" BUCKET = "site-maps-packages" API_ID = "miro8ndp7h" DIST = "E321OKRKNI9DKX" # site-maps.com (BL-173 headers) LOGS = {"/aws/lambda/site-maps-worker": 7, "/aws/lambda/site-maps-api": 7, "/aws/lambda/site-maps-digest": 7, "/aws/lambda/comp-map-worker": 7} QUEUES = {"site-maps-jobs": 86400, "site-maps-jobs-dlq": 86400, "comp-map-jobs-dlq": 86400} def aws(*args): out = subprocess.run(["aws", *args, "--region", R, "--output", "json"], capture_output=True, text=True) if out.returncode: raise RuntimeError(out.stderr.strip()[:200]) return json.loads(out.stdout or "null") def main(): bad, ok = [], [] def check(cond, what): (ok if cond else bad).append(what) groups = {g["logGroupName"]: g.get("retentionInDays") for g in aws("logs", "describe-log-groups", "--log-group-name-prefix", "/aws/lambda/")["logGroups"]} for g, days in LOGS.items(): have = groups.get(g) check(have is not None and have <= days, f"log retention {g}: {have} (want <= {days} days)") # patch 440 (BL-194): the map picker's tile function, once it exists. It logs # nothing, and Lambda's own lines carry no tile address; one day is enough. if "/aws/lambda/site-maps-tiles" in groups: have = groups["/aws/lambda/site-maps-tiles"] check(have is not None and have <= 1, f"log retention /aws/lambda/site-maps-tiles: {have} (want <= 1 day)") pitr = aws("dynamodb", "describe-continuous-backups", "--table-name", TABLE) st = pitr["ContinuousBackupsDescription"]["PointInTimeRecoveryDescription"][ "PointInTimeRecoveryStatus"] check(st == "DISABLED", f"point-in-time recovery on {TABLE}: {st} " "(a restore would bring back expired rows)") ttl = aws("dynamodb", "describe-time-to-live", "--table-name", TABLE)[ "TimeToLiveDescription"] check(ttl.get("TimeToLiveStatus") == "ENABLED" and ttl.get("AttributeName") == "expires_at", f"TTL on {TABLE}: {ttl.get('TimeToLiveStatus')} {ttl.get('AttributeName')}") ver = aws("s3api", "get-bucket-versioning", "--bucket", BUCKET) or {} check(ver.get("Status") not in ("Enabled", "Suspended"), f"versioning on {BUCKET}: {ver.get('Status') or 'never enabled'} " "(a deleted package would survive as a prior version)") for s in aws("apigatewayv2", "get-stages", "--api-id", API_ID)["Items"]: check(not s.get("AccessLogSettings"), f"API access logging on stage {s['StageName']}: " f"{'on' if s.get('AccessLogSettings') else 'off'}") for name, most in QUEUES.items(): url = aws("sqs", "get-queue-url", "--queue-name", name)["QueueUrl"] ret = int(aws("sqs", "get-queue-attributes", "--queue-url", url, "--attribute-names", "MessageRetentionPeriod")[ "Attributes"]["MessageRetentionPeriod"]) check(ret <= most, f"queue {name} keeps a message {ret} s (want <= {most})") # No stored SiteMaps job row may hold an address (a row written by an old # worker or a manual fix would break the page's first claim). held, kw = 0, {} while True: page = aws("dynamodb", "scan", "--table-name", TABLE, "--projection-expression", "job_id,address,tool", *kw.get("args", [])) held += sum(1 for i in page.get("Items", []) if "address" in i and (i.get("tool") or {}).get("S") != "compmap") nxt = page.get("LastEvaluatedKey") if not nxt: break kw["args"] = ["--exclusive-start-key", json.dumps(nxt)] check(held == 0, f"SiteMaps job rows holding an address: {held}") # Phase B (patch 355): every SiteMaps job is locked, and what is stored under # it opens with no tool - the first bytes are the lock's, never a PNG, PDF or # ZIP's. Samples up to 3 files per job (a ranged read of 5 bytes each). rows, kw = [], {} while True: page = aws("dynamodb", "scan", "--table-name", TABLE, "--projection-expression", "job_id,tool,locked,bylaw_url", *kw.get("args", [])) rows += [i for i in page.get("Items", []) if (i.get("tool") or {}).get("S") == "sitemaps"] nxt = page.get("LastEvaluatedKey") if not nxt: break kw["args"] = ["--exclusive-start-key", json.dumps(nxt)] unlocked = [i["job_id"]["S"] for i in rows if not (i.get("locked") or {}).get("BOOL")] check(not unlocked, f"SiteMaps jobs not locked: {len(unlocked)} {unlocked[:5]}") plain_bylaw = [i["job_id"]["S"] for i in rows if "bylaw_url" in i] check(not plain_bylaw, f"SiteMaps rows holding a plain by-law link (patch 355b): " f"{len(plain_bylaw)} {plain_bylaw[:5]}") plain, sampled = [], 0 with tempfile.TemporaryDirectory() as td: for i in rows: jid = i["job_id"]["S"] objs = (aws("s3api", "list-objects-v2", "--bucket", BUCKET, "--prefix", f"packages/{jid}/", "--max-keys", "40") or {}).get("Contents") or [] for o in objs[:3]: out = pathlib.Path(td) / "head" aws("s3api", "get-object", "--bucket", BUCKET, "--key", o["Key"], "--range", "bytes=0-4", str(out)) sampled += 1 if out.read_bytes()[:5] != b"SMLK1": plain.append(o["Key"]) check(not plain, f"stored SiteMaps files not locked: {len(plain)} of {sampled} sampled " f"{plain[:3]}") cors = aws("s3api", "get-bucket-cors", "--bucket", BUCKET)["CORSRules"] check(any("https://site-maps.com" in r.get("AllowedOrigins", []) for r in cors), "the package bucket lets site-maps.com fetch locked files (CORS)") sel = aws("cloudtrail", "get-event-selectors", "--trail-name", "comp-maps-audit")[ "AdvancedEventSelectors"] audited = [x for x in sel if any( "arn:aws:s3:::site-maps-packages/packages/" in f.get("StartsWith", []) for f in x["FieldSelectors"])] check(len(audited) == 1 and any( f.get("Field") == "userIdentity.arn" and f.get("NotStartsWith") for f in audited[0]["FieldSelectors"]), "package reads audited, users' reads through their links not (no user IPs kept)") # The code half: what is written where. Static, so it runs before a deploy. api = (HERE / "api_handler.py").read_text(encoding="utf-8") post = api[api.index("def _post_job("):api.index("def _sitemaps_access(")] put = post[post.index("put_item("):post.index("_sqs.send_message")] check('"address"' not in put, "SiteMaps job row written without an address") check("expires_at" in put, "SiteMaps job row written with expires_at") bpost = api[api.index("def _post_batch("):api.index("def _get_batch(")] # patch 363 bputs = bpost[bpost.index("put_item("):bpost.index("_sqs.send_message")] check('"address"' not in bputs and bputs.count("expires_at") >= 2, "batch and child rows written without inputs, with expires_at (BL-169)") dig = (HERE / "digest_handler.py").read_text(encoding="utf-8") handler = dig[dig.index("def handler("):] check(not re.search(r"""\.get\(\s*["']address["']""", handler), "digest reads no address") check(not re.search(r"""i\[\s*["']email["']\s*\]|%s.*email""", handler), "digest prints no requester email") wk = (HERE / "lambda_handler.py").read_text(encoding="utf-8") check("redirect_stdout" in wk and "00_run_log.txt" in wk, "worker console captured into the package, not CloudWatch") check("Address:" not in wk, "done-email carries no address") put = wk[wk.index("def _put("):wk.index("\ndef ", wk.index("def _put(") + 1)] check(wk.count("upload_file(") == put.count("upload_file(") == 2 and "_lock_file(" in put, "worker stores files only through _put, which locks them") page = (HERE / "site" / "p.js").read_text(encoding="utf-8") # BL-173: script moved out of p.html check("sitemaps-enc:" in page and "SMLK1" in page and "sitemaps-enc:" in wk, "page and worker agree on the lock (key label and format)") priv = (HERE / "site" / "privacy.html").read_text(encoding="utf-8") check("the operator could open it" not in priv and "stored locked" in priv, "privacy page says the package is locked") # BL-173 (2026-09-28): the security contact file must not lapse (its Expires # date is required, a year at most); fail 30 days before it does. import datetime # noqa: PLC0415 sec = HERE / "site" / ".well-known" / "security.txt" exp = re.search(r"^Expires: (\S+)", sec.read_text(encoding="utf-8"), re.MULTILINE) if sec.exists() else None check(bool(exp) and (datetime.datetime.fromisoformat(exp.group(1).replace("Z", "+00:00")) - datetime.datetime.now(datetime.timezone.utc)).days > 30, "security.txt present and more than 30 days from expiry") check('id="reporting"' in priv and "security@site-maps.com" in priv, "privacy page names the security contact") # BL-173 step 1: the code the privacy page publishes is the code in this folder. sys.path.insert(0, str(HERE)) from make_public_source import build # noqa: PLC0415 pub = HERE / "site" / "source" check(all((pub / n).exists() and (pub / n).read_text(encoding="utf-8") == t for n, t in build().items()), "published code (site/source) matches the code here - run make_public_source.py") # BL-173 step 3: the site sends the security headers in aws/cf-response-headers-policy.json. want = json.loads((HERE / "aws" / "cf-response-headers-policy.json").read_text(encoding="utf-8")) dist = aws("cloudfront", "get-distribution-config", "--id", DIST)["DistributionConfig"] pid = dist["DefaultCacheBehavior"].get("ResponseHeadersPolicyId") live = (aws("cloudfront", "get-response-headers-policy", "--id", pid) ["ResponseHeadersPolicy"]["ResponseHeadersPolicyConfig"] if pid else {}) sh, wsh = live.get("SecurityHeadersConfig", {}), want["SecurityHeadersConfig"] check(bool(pid) and sh.get("ContentSecurityPolicy", {}).get("ContentSecurityPolicy") == wsh["ContentSecurityPolicy"]["ContentSecurityPolicy"] and sh.get("ReferrerPolicy", {}).get("ReferrerPolicy") == "no-referrer" and sh.get("StrictTransportSecurity", {}).get("AccessControlMaxAgeSec", 0) >= 31536000 and live.get("CustomHeadersConfig", {}).get("Items", []) == want["CustomHeadersConfig"]["Items"], "site carries the security headers in aws/cf-response-headers-policy.json") # patch 440 (BL-194): a map-picker tile address names an area, so the site # keeps no CloudFront access log, and the tile function prints nothing and # reads nothing of the visitor's (its requests to the Province carry our # User-Agent only). check(not (dist.get("Logging") or {}).get("Enabled"), "site-maps.com keeps no CloudFront access log (a tile address names an area)") tpx = (HERE / "tile_proxy.py").read_text(encoding="utf-8") tcode = re.sub(r'"""[\s\S]*?"""', "", "\n".join(l for l in tpx.splitlines() if not l.lstrip().startswith("#"))) check("print(" not in tcode and "logging" not in tcode and "forwarded" not in tcode.lower(), "tile function prints nothing and reads no visitor header (BL-194)") # patch 409b (IP-25; IP Issues and CompMaps, 2026-10-04): the shared tile cache # keeps zoom 14 and wider only. A closer tile would record where a package was # drawn. Checked on both writers: the constant here, and the live prefix (as # CompMaps' own check does, whichever product wrote the object). The listing # runs as the operator; the bucket policy denies it to every role (CM-41). src = (HERE.parent / "sitemaps.py").read_text(encoding="utf-8") check(re.search(r"^_TILE_CACHE_MAX_Z = 14\b", src, re.MULTILINE) is not None, "SiteMaps caches tiles at zoom 14 and wider only (_TILE_CACHE_MAX_Z = 14)") keys = aws("s3api", "list-objects-v2", "--bucket", BUCKET, "--prefix", "tilecache/osm/", "--query", "Contents[].Key") or [] close = [k for k in keys if k.count("/") >= 4 and k.split("/")[2].isdigit() and int(k.split("/")[2]) > 14] check(not close, f"no tile closer than zoom 14 under tilecache/osm/ ({len(close)} found)") cache_on = aws("lambda", "get-function-configuration", "--function-name", "site-maps-worker", "--query", "Environment.Variables.PR_TILE_CACHE") check(cache_on != "s3" or "Map tiles" in priv, "privacy page has its map-tiles line before the worker caches tiles") for w in ok: print(" ok " + w) for w in bad: print(" FAIL " + w) print("privacy_check: %d ok, %d failing" % (len(ok), len(bad))) return 1 if bad else 0 if __name__ == "__main__": sys.exit(main())