#!/usr/bin/env python3 """identity-resolve.py — Pure identity resolution for the identity plane. Reads identity-map.json (fingerprints only, never key material) and resolves an API-key fingerprint to its network-identity scope: api_key -> account_origin(s); one origin rolls scope UP to the umbrella account, two or more keep scope DOWN at the key itself. This module is pure + total (missing/corrupt map -> empty, unknown fingerprint -> None). CLI output carries emails and fingerprints only; key bytes never appear here — there is no code path that reads them except `fp`, which hashes stdin and prints only the digest. Usage: identity-resolve.py fp < keyfile # print sha256: fingerprint identity-resolve.py lookup # print scope JSON identity-resolve.py check # validate map schema """ from __future__ import annotations import argparse import hashlib import json import re import sys from pathlib import Path from typing import Any, Dict, List, Optional REPO_ROOT = Path(__file__).resolve().parent.parent MAP_FILE = REPO_ROOT / "identity-map.json" LABEL_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,22}$") def fingerprint_hex(material: bytes) -> str: """sha256: fingerprint of raw key bytes.""" return "sha256:" + hashlib.sha256(material).hexdigest() def load_map(path: str | Path = MAP_FILE) -> Dict[str, Any]: """Load the identity map. Missing/corrupt -> {"accounts": {}}.""" try: with open(path, "r") as f: data = json.load(f) if isinstance(data, dict) and isinstance( data.get("accounts"), dict): return data except Exception: pass return {"accounts": {}} def find_key(map_data: Dict[str, Any], fp: str) -> Optional[Dict[str, Any]]: """Locate a key record by fingerprint. Returns {"email", "key"} or None. Top-level '_' entries ignored. """ if not fp: return None accounts = map_data.get("accounts") if not isinstance(accounts, dict): return None for email, rec in accounts.items(): if not isinstance(rec, dict): continue keys = rec.get("keys") if not isinstance(keys, list): continue for k in keys: if isinstance(k, dict) and k.get("fp") == fp: return {"email": email, "key": k} return None def resolve_scope(map_data: Dict[str, Any], fp: str) -> Optional[Dict[str, Any]]: """Resolve a fingerprint to its scope unit. Single origin -> {"scope": "account", "unit": email, ...}. Multiple origins -> {"scope": "key", "unit": fp, ...}. Unknown fingerprint -> None. Result carries emails + fingerprints only (no key material exists anywhere in this module). """ found = find_key(map_data, fp) if found is None: return None key = found["key"] origins = key.get("origins") if not isinstance(origins, list) or not origins: return None origins = [str(o) for o in origins] if len(origins) == 1: return {"scope": "account", "unit": origins[0], "email": found["email"], "origins": origins, "label": key.get("label", "")} return {"scope": "key", "unit": fp, "email": found["email"], "origins": origins, "label": key.get("label", "")} def scope_slug(scope: Dict[str, Any]) -> str: """Deterministic netvm label for a scope (fits label validation). Account scopes: id--. Key scopes: id-k-. Always matches ^[a-z0-9][a-z0-9-]{0,22}$. """ unit = str(scope.get("unit", "")) if scope.get("scope") == "key": hexpart = re.sub(r"[^0-9a-f]", "", unit.lower())[:12] or "0" return "id-k-%s" % hexpart frag = re.sub(r"[^a-z0-9]+", "-", unit.lower()).strip("-")[:12] frag = frag.strip("-") or "x" tag = hashlib.sha256(unit.encode()).hexdigest()[:7] return "id-%s-%s" % (frag, tag) def check_map(map_data: Dict[str, Any]) -> List[str]: """Validate map schema. Returns a list of problem strings (empty OK).""" problems: List[str] = [] accounts = map_data.get("accounts") if not isinstance(accounts, dict): return ["top-level 'accounts' must be an object"] seen_fps: Dict[str, str] = {} for email, rec in accounts.items(): if not isinstance(email, str) or "@" not in email: problems.append("account key %r is not an email" % (email,)) if not isinstance(rec, dict) or not isinstance( rec.get("keys"), list): problems.append("account %r: 'keys' must be a list" % (email,)) continue for i, k in enumerate(rec["keys"]): where = "%s.keys[%d]" % (email, i) if not isinstance(k, dict): problems.append("%s: not an object" % where) continue fp = k.get("fp", "") if not re.fullmatch(r"sha256:[0-9a-f]{64}", str(fp)): problems.append("%s: bad fingerprint %r" % (where, fp)) elif fp in seen_fps: problems.append("%s: fingerprint already listed under %s" % (where, seen_fps[fp])) else: seen_fps[fp] = email origins = k.get("origins") if not isinstance(origins, list) or not origins or not all( isinstance(o, str) and o for o in origins): problems.append("%s: 'origins' must be a non-empty " "string list" % where) return problems def main(argv: Optional[List[str]] = None) -> int: ap = argparse.ArgumentParser(prog="identity-resolve.py") ap.add_argument("--map", default=str(MAP_FILE), help="identity map (default: identity-map.json)") sub = ap.add_subparsers(dest="cmd", required=True) sub.add_parser("fp", help="print sha256: fingerprint of stdin bytes") p = sub.add_parser("lookup", help="resolve a fingerprint to scope JSON") p.add_argument("fp") sub.add_parser("check", help="validate the map schema") args = ap.parse_args(argv) if args.cmd == "fp": sys.stdout.write(fingerprint_hex(sys.stdin.buffer.read()) + "\n") return 0 if args.cmd == "lookup": scope = resolve_scope(load_map(args.map), args.fp) if scope is None: print("unknown fingerprint (not in map)") return 1 scope["slug"] = scope_slug(scope) print(json.dumps(scope, indent=2)) return 0 problems = check_map(load_map(args.map)) if problems: print("%s INVALID:" % args.map) for prob in problems: print(" - %s" % prob) return 1 print("%s OK" % args.map) return 0 if __name__ == "__main__": sys.exit(main())