edge_gateway_admin.py 52 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323
  1. #!/usr/bin/env python3
  2. """Fail-closed local administration for a DataOps enterprise edge gateway."""
  3. from __future__ import annotations
  4. import argparse
  5. import fcntl
  6. import hashlib
  7. import json
  8. import os
  9. import re
  10. import sqlite3
  11. import stat
  12. import sys
  13. import tempfile
  14. from contextlib import contextmanager, suppress
  15. from datetime import UTC, datetime
  16. from decimal import Decimal
  17. from pathlib import Path, PurePosixPath
  18. from cryptography import x509
  19. from cryptography.exceptions import InvalidSignature
  20. from cryptography.hazmat.primitives import hashes, serialization
  21. from cryptography.hazmat.primitives.asymmetric import (
  22. ec,
  23. ed448,
  24. ed25519,
  25. padding,
  26. rsa,
  27. )
  28. from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID
  29. MAX_JSON_BYTES = 262_144
  30. MAX_BACKUPS = 3
  31. SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
  32. IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,254}$")
  33. VERSION_RE = re.compile(r"^(0|[1-9][0-9]{0,9})\.(0|[1-9][0-9]{0,9})\.(0|[1-9][0-9]{0,9})$")
  34. ORIGIN_RE = re.compile(r"^https://[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?(?::[0-9]{1,5})?$")
  35. MANIFEST_FIELDS = frozenset(
  36. {
  37. "release_id", "version", "rollback_version", "artifact_digest",
  38. "artifact_name", "deadline_at", "signature_algorithm", "key_id",
  39. "manifest_digest", "signature", "status",
  40. }
  41. )
  42. class AdminError(RuntimeError):
  43. def __init__(self, code: str, exit_code: int = 2):
  44. super().__init__(code)
  45. self.code = code
  46. self.exit_code = exit_code
  47. def _json_bytes(value: object) -> bytes:
  48. return json.dumps(
  49. value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
  50. ).encode("utf-8")
  51. def _digest(value: object) -> str:
  52. return hashlib.sha256(_json_bytes(value)).hexdigest()
  53. def _protocol_bytes(value: object) -> bytes:
  54. def frame(tag: bytes, content: bytes) -> bytes:
  55. return tag + str(len(content)).encode("ascii") + b":" + content
  56. if value is None:
  57. return b"n"
  58. if isinstance(value, bool):
  59. return b"b1" if value else b"b0"
  60. if isinstance(value, (int, float)):
  61. number = Decimal(value if isinstance(value, int) else str(value))
  62. if not number.is_finite():
  63. raise AdminError("INVALID_INPUT")
  64. text = "0" if number == 0 else format(number.normalize(), "f")
  65. if "." in text:
  66. text = text.rstrip("0").rstrip(".")
  67. return frame(b"d", text.encode("ascii"))
  68. if isinstance(value, str):
  69. return frame(b"s", value.encode("utf-8"))
  70. if isinstance(value, dict):
  71. if any(not isinstance(key, str) for key in value):
  72. raise AdminError("INVALID_INPUT")
  73. encoded = bytearray(b"m" + str(len(value)).encode("ascii") + b":")
  74. for key in sorted(value, key=lambda item: item.encode("utf-8")):
  75. encoded.extend(frame(b"k", key.encode("utf-8")))
  76. encoded.extend(_protocol_bytes(value[key]))
  77. return bytes(encoded)
  78. if isinstance(value, (list, tuple)):
  79. encoded = bytearray(b"l" + str(len(value)).encode("ascii") + b":")
  80. for item in value:
  81. encoded.extend(_protocol_bytes(item))
  82. return bytes(encoded)
  83. raise AdminError("INVALID_INPUT")
  84. def _protocol_digest(value: object) -> str:
  85. return hashlib.sha256(_protocol_bytes(value)).hexdigest()
  86. def _output(value: dict[str, object]) -> None:
  87. rendered = json.dumps(value, ensure_ascii=False, sort_keys=True)
  88. if len(rendered.encode("utf-8")) > 32_768:
  89. raise AdminError("OUTPUT_LIMIT")
  90. print(rendered)
  91. def _root(raw: str, *, create: bool = False) -> Path:
  92. candidate = Path(raw)
  93. if not candidate.is_absolute() or "\x00" in raw:
  94. raise AdminError("INVALID_INPUT")
  95. protected = {
  96. Path("/"), Path.home(), Path("/opt"), Path("/var"), Path("/usr"),
  97. Path("/etc"), Path("/tmp"), Path("/private"), Path("/private/tmp"),
  98. Path("/Users"), Path("/home"), Path("/root"), Path("/Volumes"),
  99. }
  100. normalized = Path(os.path.abspath(candidate))
  101. if normalized in protected:
  102. raise AdminError("PATH_POLICY")
  103. if candidate.exists() and candidate.is_symlink():
  104. raise AdminError("PATH_POLICY")
  105. if create:
  106. with suppress(FileExistsError):
  107. candidate.mkdir(mode=0o700, parents=False, exist_ok=False)
  108. if not candidate.is_dir():
  109. raise AdminError("PATH_POLICY")
  110. resolved = candidate.resolve(strict=True)
  111. if resolved != candidate:
  112. raise AdminError("PATH_POLICY")
  113. return resolved
  114. def _runtime_identity(root: Path) -> tuple[int, int]:
  115. record = _read_json(root / "state" / "runtime-identity.json")
  116. if set(record) != {"runtime_gid", "runtime_uid", "schema_version"}:
  117. raise AdminError("OWNER_POLICY")
  118. uid, gid = record["runtime_uid"], record["runtime_gid"]
  119. if (
  120. isinstance(uid, bool)
  121. or not isinstance(uid, int)
  122. or isinstance(gid, bool)
  123. or not isinstance(gid, int)
  124. or uid < 1
  125. or gid < 1
  126. ):
  127. raise AdminError("OWNER_POLICY")
  128. metadata = root.stat()
  129. if metadata.st_uid != uid or metadata.st_gid != gid:
  130. raise AdminError("OWNER_POLICY")
  131. return uid, gid
  132. def _validated_root(raw: str) -> Path:
  133. root = _root(raw)
  134. _runtime_identity(root)
  135. return root
  136. def _chown_exact(path: Path, uid: int, gid: int) -> None:
  137. metadata = path.lstat()
  138. if stat.S_ISLNK(metadata.st_mode):
  139. raise AdminError("PATH_POLICY")
  140. if (metadata.st_uid, metadata.st_gid) == (uid, gid):
  141. return
  142. if os.geteuid() != 0:
  143. raise AdminError("OWNER_POLICY")
  144. os.chown(path, uid, gid, follow_symlinks=False)
  145. def _inside(root: Path, raw: str | Path, *, must_exist: bool = False) -> Path:
  146. candidate = Path(raw)
  147. if not candidate.is_absolute() or "\x00" in str(candidate):
  148. raise AdminError("INVALID_INPUT")
  149. if candidate.is_symlink():
  150. raise AdminError("PATH_POLICY")
  151. try:
  152. parent = candidate.parent.resolve(strict=True)
  153. parent.relative_to(root)
  154. except (OSError, ValueError) as exc:
  155. raise AdminError("PATH_POLICY") from exc
  156. current = root
  157. for part in parent.relative_to(root).parts:
  158. current /= part
  159. if current.is_symlink():
  160. raise AdminError("PATH_POLICY")
  161. if must_exist:
  162. try:
  163. resolved = candidate.resolve(strict=True)
  164. resolved.relative_to(root)
  165. except (OSError, ValueError) as exc:
  166. raise AdminError("PATH_POLICY") from exc
  167. metadata = candidate.stat()
  168. if (
  169. candidate.is_symlink()
  170. or not stat.S_ISREG(metadata.st_mode)
  171. or metadata.st_nlink != 1
  172. ):
  173. raise AdminError("PATH_POLICY")
  174. return candidate
  175. def _mkdir(root: Path, name: str, mode: int = 0o700) -> Path:
  176. path = root / name
  177. if path.exists() and (path.is_symlink() or not path.is_dir()):
  178. raise AdminError("PATH_POLICY")
  179. created = not path.exists()
  180. path.mkdir(mode=mode, exist_ok=True)
  181. os.chmod(path, mode)
  182. parent = path.parent.stat()
  183. metadata = path.stat()
  184. if created and os.geteuid() == 0:
  185. os.chown(path, parent.st_uid, parent.st_gid, follow_symlinks=False)
  186. elif (metadata.st_uid, metadata.st_gid) != (parent.st_uid, parent.st_gid):
  187. raise AdminError("OWNER_POLICY")
  188. return path
  189. def _inherit_parent_owner(descriptor: int, parent: Path) -> None:
  190. parent_metadata = parent.stat()
  191. metadata = os.fstat(descriptor)
  192. expected = (parent_metadata.st_uid, parent_metadata.st_gid)
  193. if (metadata.st_uid, metadata.st_gid) == expected:
  194. return
  195. if os.geteuid() != 0:
  196. raise AdminError("OWNER_POLICY")
  197. os.fchown(descriptor, *expected)
  198. def _atomic_write(path: Path, data: bytes, mode: int) -> None:
  199. if path.exists() and path.is_symlink():
  200. raise AdminError("PATH_POLICY")
  201. descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
  202. try:
  203. os.fchmod(descriptor, mode)
  204. _inherit_parent_owner(descriptor, path.parent)
  205. with os.fdopen(descriptor, "wb", closefd=True) as stream:
  206. stream.write(data)
  207. stream.flush()
  208. os.fsync(stream.fileno())
  209. os.replace(temporary, path)
  210. os.chmod(path, mode)
  211. _fsync_directory(path.parent)
  212. except Exception:
  213. with suppress(FileNotFoundError):
  214. os.unlink(temporary)
  215. raise
  216. def _fsync_directory(path: Path) -> None:
  217. directory = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
  218. try:
  219. os.fsync(directory)
  220. finally:
  221. os.close(directory)
  222. def _atomic_copy(source: Path, target: Path, mode: int) -> None:
  223. if target.exists() or target.is_symlink():
  224. raise AdminError("PATH_POLICY")
  225. descriptor, temporary = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent)
  226. try:
  227. os.fchmod(descriptor, mode)
  228. _inherit_parent_owner(descriptor, target.parent)
  229. with source.open("rb") as reader, os.fdopen(descriptor, "wb") as writer:
  230. for chunk in iter(lambda: reader.read(1_048_576), b""):
  231. writer.write(chunk)
  232. writer.flush()
  233. os.fsync(writer.fileno())
  234. os.replace(temporary, target)
  235. os.chmod(target, mode)
  236. _fsync_directory(target.parent)
  237. except Exception:
  238. with suppress(FileNotFoundError):
  239. os.unlink(temporary)
  240. raise
  241. def _read_json(path: Path) -> dict[str, object]:
  242. if not path.is_file() or path.is_symlink() or path.stat().st_size > MAX_JSON_BYTES:
  243. raise AdminError("INVALID_INPUT")
  244. try:
  245. value = json.loads(path.read_text(encoding="utf-8"))
  246. except (OSError, UnicodeError, json.JSONDecodeError) as exc:
  247. raise AdminError("INVALID_INPUT") from exc
  248. if not isinstance(value, dict):
  249. raise AdminError("INVALID_INPUT")
  250. return value
  251. def _sha256_file(path: Path) -> str:
  252. digest = hashlib.sha256()
  253. try:
  254. with path.open("rb") as stream:
  255. for chunk in iter(lambda: stream.read(1_048_576), b""):
  256. digest.update(chunk)
  257. except OSError as exc:
  258. raise AdminError("INVALID_INPUT") from exc
  259. return digest.hexdigest()
  260. def _open_root_file(root: Path, raw: str | Path) -> tuple[int, os.stat_result]:
  261. path = _inside(root, raw)
  262. flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
  263. if hasattr(os, "O_NOFOLLOW"):
  264. flags |= os.O_NOFOLLOW
  265. try:
  266. descriptor = os.open(path, flags)
  267. except OSError as exc:
  268. raise AdminError("PATH_POLICY") from exc
  269. try:
  270. metadata = os.fstat(descriptor)
  271. uid, gid = _runtime_identity(root)
  272. if (
  273. not stat.S_ISREG(metadata.st_mode)
  274. or metadata.st_nlink != 1
  275. or (metadata.st_uid, metadata.st_gid) != (uid, gid)
  276. ):
  277. raise AdminError("PATH_POLICY")
  278. return descriptor, metadata
  279. except Exception:
  280. os.close(descriptor)
  281. raise
  282. def _same_file_snapshot(before: os.stat_result, after: os.stat_result) -> bool:
  283. """Compare mutation-relevant descriptor metadata without treating atime as a write."""
  284. fields = (
  285. "st_dev", "st_ino", "st_mode", "st_nlink", "st_uid", "st_gid",
  286. "st_size", "st_mtime_ns", "st_ctime_ns",
  287. )
  288. return all(getattr(before, field) == getattr(after, field) for field in fields)
  289. def _read_json_descriptor(descriptor: int) -> dict[str, object]:
  290. metadata = os.fstat(descriptor)
  291. if metadata.st_size > MAX_JSON_BYTES:
  292. raise AdminError("INVALID_INPUT")
  293. os.lseek(descriptor, 0, os.SEEK_SET)
  294. try:
  295. raw = b""
  296. while len(raw) <= MAX_JSON_BYTES:
  297. chunk = os.read(descriptor, min(65_536, MAX_JSON_BYTES + 1 - len(raw)))
  298. if not chunk:
  299. break
  300. raw += chunk
  301. value = json.loads(raw.decode("utf-8"))
  302. except (OSError, UnicodeError, json.JSONDecodeError) as exc:
  303. raise AdminError("INVALID_INPUT") from exc
  304. if not isinstance(value, dict):
  305. raise AdminError("INVALID_INPUT")
  306. return value
  307. def _copy_descriptor(
  308. source_descriptor: int, source_before: os.stat_result, target: Path
  309. ) -> str:
  310. flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0)
  311. if hasattr(os, "O_NOFOLLOW"):
  312. flags |= os.O_NOFOLLOW
  313. target_descriptor = os.open(target, flags, 0o600)
  314. digest = hashlib.sha256()
  315. try:
  316. os.fchmod(target_descriptor, 0o600)
  317. _inherit_parent_owner(target_descriptor, target.parent)
  318. os.lseek(source_descriptor, 0, os.SEEK_SET)
  319. while True:
  320. chunk = os.read(source_descriptor, 1_048_576)
  321. if not chunk:
  322. break
  323. digest.update(chunk)
  324. view = memoryview(chunk)
  325. while view:
  326. written = os.write(target_descriptor, view)
  327. view = view[written:]
  328. os.fsync(target_descriptor)
  329. finally:
  330. os.close(target_descriptor)
  331. source_after = os.fstat(source_descriptor)
  332. snapshot_fields = (
  333. "st_dev", "st_ino", "st_mode", "st_nlink", "st_uid", "st_gid",
  334. "st_size", "st_mtime_ns", "st_ctime_ns",
  335. )
  336. if any(
  337. getattr(source_before, field) != getattr(source_after, field)
  338. for field in snapshot_fields
  339. ):
  340. raise AdminError("RELEASE_REJECTED", 3)
  341. copied_digest = digest.hexdigest()
  342. if _sha256_file(target) != copied_digest:
  343. raise AdminError("RELEASE_REJECTED", 3)
  344. return copied_digest
  345. def _layout(root: Path) -> None:
  346. for name in (
  347. "queue", "artifacts", "releases", "secrets", "certificates",
  348. "registration", "state",
  349. ):
  350. _mkdir(root, name)
  351. @contextmanager
  352. def _process_lock(root: Path, uid: int | None = None, gid: int | None = None):
  353. if uid is None or gid is None:
  354. uid, gid = _runtime_identity(root)
  355. lock_path = root / "state" / "admin.lock"
  356. flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_CLOEXEC", 0)
  357. if hasattr(os, "O_NOFOLLOW"):
  358. flags |= os.O_NOFOLLOW
  359. try:
  360. descriptor = os.open(lock_path, flags, 0o600)
  361. except OSError as exc:
  362. raise AdminError("LOCK_POLICY", 3) from exc
  363. try:
  364. if os.geteuid() == 0:
  365. os.fchown(descriptor, uid, gid)
  366. os.fchmod(descriptor, 0o600)
  367. metadata = os.fstat(descriptor)
  368. if (
  369. not stat.S_ISREG(metadata.st_mode)
  370. or stat.S_IMODE(metadata.st_mode) != 0o600
  371. or metadata.st_nlink != 1
  372. or (metadata.st_uid, metadata.st_gid) != (uid, gid)
  373. ):
  374. raise AdminError("LOCK_POLICY", 3)
  375. try:
  376. fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
  377. except BlockingIOError as exc:
  378. raise AdminError("CONCURRENT_OPERATION", 3) from exc
  379. yield
  380. finally:
  381. os.close(descriptor)
  382. def _bounded_backups(directory: Path, prefix: str) -> None:
  383. candidates = sorted(
  384. (path for path in directory.glob(f"{prefix}*") if path.is_file() and not path.is_symlink()),
  385. key=lambda path: path.stat().st_mtime_ns,
  386. reverse=True,
  387. )
  388. for path in candidates[MAX_BACKUPS:]:
  389. path.unlink()
  390. def command_init(args: argparse.Namespace) -> dict[str, object]:
  391. if (
  392. isinstance(args.runtime_uid, bool)
  393. or not 1 <= args.runtime_uid <= 2_147_483_647
  394. or isinstance(args.runtime_gid, bool)
  395. or not 1 <= args.runtime_gid <= 2_147_483_647
  396. ):
  397. raise AdminError("INVALID_INPUT")
  398. root = _root(args.root, create=True)
  399. identity_path = root / "state" / "runtime-identity.json"
  400. if identity_path.exists():
  401. uid, gid = _runtime_identity(root)
  402. if (uid, gid) != (args.runtime_uid, args.runtime_gid):
  403. raise AdminError("OWNER_POLICY")
  404. target = root / "edge-config.template.json"
  405. if not target.is_file() or target.is_symlink():
  406. raise AdminError("PATH_POLICY")
  407. with _process_lock(root):
  408. return {
  409. "config_template_path": str(target),
  410. "root": str(root),
  411. "runtime_gid": gid,
  412. "runtime_uid": uid,
  413. "status": "initialized",
  414. }
  415. if any(root.iterdir()):
  416. raise AdminError("PATH_POLICY")
  417. os.chmod(root, 0o700)
  418. current = root.stat()
  419. if (current.st_uid, current.st_gid) != (args.runtime_uid, args.runtime_gid):
  420. if not args.allow_chown:
  421. raise AdminError("OWNER_POLICY")
  422. _chown_exact(root, args.runtime_uid, args.runtime_gid)
  423. _layout(root)
  424. for name in (
  425. "queue", "artifacts", "releases", "secrets", "certificates",
  426. "registration", "state",
  427. ):
  428. _chown_exact(root / name, args.runtime_uid, args.runtime_gid)
  429. identity = {
  430. "runtime_gid": args.runtime_gid,
  431. "runtime_uid": args.runtime_uid,
  432. "schema_version": 1,
  433. }
  434. with _process_lock(root, args.runtime_uid, args.runtime_gid):
  435. _atomic_write(identity_path, _json_bytes(identity) + b"\n", 0o600)
  436. _chown_exact(identity_path, args.runtime_uid, args.runtime_gid)
  437. template = {
  438. "allowed_control_hosts": [],
  439. "allowed_proxy_hosts": [],
  440. "control_origin": None,
  441. "environment": None,
  442. "gateway_id": None,
  443. "network_zone": None,
  444. "policy_digest": None,
  445. "proxy_origin": None,
  446. "runtime_gid": args.runtime_gid,
  447. "runtime_uid": args.runtime_uid,
  448. "schema_version": 1,
  449. }
  450. target = root / "edge-config.template.json"
  451. _atomic_write(target, _json_bytes(template) + b"\n", 0o600)
  452. _chown_exact(target, args.runtime_uid, args.runtime_gid)
  453. return {
  454. "config_template_path": str(target),
  455. "root": str(root),
  456. "runtime_gid": args.runtime_gid,
  457. "runtime_uid": args.runtime_uid,
  458. "status": "initialized",
  459. }
  460. def _new_csr(root: Path, common_name: str, prefix: str) -> dict[str, object]:
  461. if not IDENTIFIER_RE.fullmatch(common_name):
  462. raise AdminError("INVALID_INPUT")
  463. key_path = root / "secrets" / f"{prefix}.key"
  464. csr_path = root / "certificates" / f"{prefix}.csr"
  465. if key_path.exists() or csr_path.exists():
  466. raise AdminError("MATERIAL_EXISTS")
  467. private = rsa.generate_private_key(public_exponent=65537, key_size=3072)
  468. csr = (
  469. x509.CertificateSigningRequestBuilder()
  470. .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)]))
  471. .sign(private, hashes.SHA256())
  472. )
  473. key_bytes = private.private_bytes(
  474. serialization.Encoding.PEM,
  475. serialization.PrivateFormat.PKCS8,
  476. serialization.NoEncryption(),
  477. )
  478. csr_bytes = csr.public_bytes(serialization.Encoding.PEM)
  479. try:
  480. _atomic_write(key_path, key_bytes, 0o600)
  481. _atomic_write(csr_path, csr_bytes, 0o644)
  482. except Exception:
  483. if key_path.exists():
  484. key_path.unlink()
  485. if csr_path.exists():
  486. csr_path.unlink()
  487. raise
  488. return {
  489. "csr_path": str(csr_path),
  490. "private_key_path": str(key_path),
  491. "public_key_sha256": hashlib.sha256(
  492. csr.public_key().public_bytes(
  493. serialization.Encoding.DER,
  494. serialization.PublicFormat.SubjectPublicKeyInfo,
  495. )
  496. ).hexdigest(),
  497. "status": "csr_created",
  498. }
  499. def command_csr(args: argparse.Namespace) -> dict[str, object]:
  500. root = _validated_root(args.root)
  501. _layout(root)
  502. return _new_csr(root, args.common_name, "client")
  503. def _verify_x509_signature(value, issuer_public_key) -> None:
  504. try:
  505. if isinstance(issuer_public_key, rsa.RSAPublicKey):
  506. issuer_public_key.verify(
  507. value.signature, value.tbs_certificate_bytes
  508. if isinstance(value, x509.Certificate)
  509. else value.tbs_certlist_bytes,
  510. padding.PKCS1v15(), value.signature_hash_algorithm,
  511. )
  512. elif isinstance(issuer_public_key, ec.EllipticCurvePublicKey):
  513. issuer_public_key.verify(
  514. value.signature, value.tbs_certificate_bytes
  515. if isinstance(value, x509.Certificate)
  516. else value.tbs_certlist_bytes,
  517. ec.ECDSA(value.signature_hash_algorithm),
  518. )
  519. elif isinstance(issuer_public_key, (ed25519.Ed25519PublicKey, ed448.Ed448PublicKey)):
  520. issuer_public_key.verify(
  521. value.signature, value.tbs_certificate_bytes
  522. if isinstance(value, x509.Certificate)
  523. else value.tbs_certlist_bytes,
  524. )
  525. else:
  526. raise AdminError("MATERIAL_INVALID")
  527. except (InvalidSignature, ValueError, TypeError) as exc:
  528. raise AdminError("MATERIAL_INVALID") from exc
  529. def _load_material(
  530. root: Path, certificate_raw: str, ca_raw: str, crl_raw: str,
  531. prefix: str = "client",
  532. ):
  533. certificate_path = _inside(root, certificate_raw, must_exist=True)
  534. ca_path = _inside(root, ca_raw, must_exist=True)
  535. crl_path = _inside(root, crl_raw, must_exist=True)
  536. key_path = _inside(root, root / "secrets" / f"{prefix}.key", must_exist=True)
  537. csr_path = _inside(root, root / "certificates" / f"{prefix}.csr", must_exist=True)
  538. if stat.S_IMODE(key_path.stat().st_mode) & 0o077:
  539. raise AdminError("KEY_PERMISSION")
  540. try:
  541. certificate = x509.load_pem_x509_certificate(certificate_path.read_bytes())
  542. authorities = x509.load_pem_x509_certificates(ca_path.read_bytes())
  543. crl = x509.load_pem_x509_crl(crl_path.read_bytes())
  544. private = serialization.load_pem_private_key(key_path.read_bytes(), password=None)
  545. csr = x509.load_pem_x509_csr(csr_path.read_bytes())
  546. except (ValueError, TypeError, OSError) as exc:
  547. raise AdminError("MATERIAL_INVALID") from exc
  548. public = certificate.public_key().public_bytes(
  549. serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
  550. )
  551. if public != private.public_key().public_bytes(
  552. serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
  553. ) or public != csr.public_key().public_bytes(
  554. serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
  555. ) or not csr.is_signature_valid:
  556. raise AdminError("MATERIAL_MISMATCH")
  557. now = datetime.now(UTC)
  558. if certificate.not_valid_before_utc > now or certificate.not_valid_after_utc <= now:
  559. raise AdminError("MATERIAL_INVALID")
  560. try:
  561. usages = certificate.extensions.get_extension_for_class(
  562. x509.ExtendedKeyUsage
  563. ).value
  564. except x509.ExtensionNotFound as exc:
  565. raise AdminError("MATERIAL_INVALID") from exc
  566. if ExtendedKeyUsageOID.CLIENT_AUTH not in usages:
  567. raise AdminError("MATERIAL_INVALID")
  568. issuer = next(
  569. (authority for authority in authorities if authority.subject == certificate.issuer),
  570. None,
  571. )
  572. if issuer is None:
  573. raise AdminError("MATERIAL_INVALID")
  574. try:
  575. basic = issuer.extensions.get_extension_for_class(x509.BasicConstraints).value
  576. except x509.ExtensionNotFound as exc:
  577. raise AdminError("MATERIAL_INVALID") from exc
  578. if not basic.ca or issuer.not_valid_after_utc <= now:
  579. raise AdminError("MATERIAL_INVALID")
  580. _verify_x509_signature(certificate, issuer.public_key())
  581. if crl.issuer != issuer.subject or crl.last_update_utc > now or crl.next_update_utc <= now:
  582. raise AdminError("MATERIAL_INVALID")
  583. _verify_x509_signature(crl, issuer.public_key())
  584. if crl.get_revoked_certificate_by_serial_number(certificate.serial_number) is not None:
  585. raise AdminError("MATERIAL_INVALID")
  586. return certificate, csr, ca_path, crl, crl_path
  587. def command_register_manifest(args: argparse.Namespace) -> dict[str, object]:
  588. root = _validated_root(args.root)
  589. certificate, csr, ca_path, crl, crl_path = _load_material(
  590. root, args.certificate, args.ca_bundle, args.crl
  591. )
  592. for value in (args.gateway_id, args.environment, args.network_zone):
  593. if not IDENTIFIER_RE.fullmatch(value):
  594. raise AdminError("INVALID_INPUT")
  595. if args.environment not in {"development", "staging", "production"}:
  596. raise AdminError("INVALID_INPUT")
  597. if not SHA256_RE.fullmatch(args.policy_digest):
  598. raise AdminError("INVALID_INPUT")
  599. if not ORIGIN_RE.fullmatch(args.control_origin):
  600. raise AdminError("INVALID_INPUT")
  601. if args.proxy_origin and not ORIGIN_RE.fullmatch(args.proxy_origin):
  602. raise AdminError("INVALID_INPUT")
  603. output = root / "registration" / "registration.json"
  604. if output.is_symlink():
  605. raise AdminError("PATH_POLICY")
  606. certificate_digest = certificate.fingerprint(hashes.SHA256()).hex()
  607. public_digest = hashlib.sha256(
  608. certificate.public_key().public_bytes(
  609. serialization.Encoding.DER,
  610. serialization.PublicFormat.SubjectPublicKeyInfo,
  611. )
  612. ).hexdigest()
  613. material = {
  614. "allowed_control_hosts": [args.control_origin.removeprefix("https://").split(":", 1)[0]],
  615. "allowed_proxy_hosts": [] if not args.proxy_origin else [args.proxy_origin.removeprefix("https://").split(":", 1)[0]],
  616. "certificate_sha256": certificate_digest,
  617. "certificate_not_after": certificate.not_valid_after_utc.isoformat().replace("+00:00", "Z"),
  618. "certificate_not_before": certificate.not_valid_before_utc.isoformat().replace("+00:00", "Z"),
  619. "control_origin": args.control_origin,
  620. "csr_sha256": hashlib.sha256(
  621. csr.public_bytes(serialization.Encoding.DER)
  622. ).hexdigest(),
  623. "public_key_sha256": public_digest,
  624. "environment": args.environment,
  625. "gateway_id": args.gateway_id,
  626. "network_zone": args.network_zone,
  627. "policy_digest": args.policy_digest,
  628. "proxy_origin": args.proxy_origin,
  629. "ca_bundle_sha256": _sha256_file(ca_path),
  630. "ca_chain_verified": True,
  631. "client_auth_eku_verified": True,
  632. "crl_sha256": _sha256_file(crl_path),
  633. "crl_verified": True,
  634. "crl_this_update": crl.last_update_utc.isoformat().replace("+00:00", "Z"),
  635. "crl_next_update": crl.next_update_utc.isoformat().replace("+00:00", "Z"),
  636. "schema_version": 1,
  637. }
  638. replayed = False
  639. if output.exists():
  640. if _read_json(output) != material:
  641. raise AdminError("IDEMPOTENCY_CONFLICT", 3)
  642. replayed = True
  643. else:
  644. _atomic_write(output, _json_bytes(material) + b"\n", 0o600)
  645. return {
  646. "certificate_sha256": certificate_digest,
  647. "manifest_digest": _digest(material),
  648. "manifest_path": str(output),
  649. "replayed": replayed,
  650. "status": "registration_material_ready",
  651. }
  652. def command_rotate(args: argparse.Namespace) -> dict[str, object]:
  653. root = _validated_root(args.root)
  654. if not IDENTIFIER_RE.fullmatch(args.request_id) or not SHA256_RE.fullmatch(
  655. args.expected_current_certificate_digest
  656. ):
  657. raise AdminError("INVALID_INPUT")
  658. request = {
  659. "common_name": args.common_name,
  660. "expected_current_certificate_digest": args.expected_current_certificate_digest,
  661. "request_id": args.request_id,
  662. }
  663. request_digest = _digest(request)
  664. record_path = root / "state" / f"rotation-{args.request_id}.json"
  665. if record_path.exists():
  666. record = _read_json(record_path)
  667. if record.get("request_digest") != request_digest:
  668. raise AdminError("IDEMPOTENCY_CONFLICT", 3)
  669. return record
  670. pending_records = [
  671. path
  672. for path in (root / "state").glob("rotation-*.json")
  673. if path.is_file()
  674. and not path.is_symlink()
  675. and _read_json(path).get("status") == "pending_server_approval"
  676. ]
  677. if len(pending_records) >= MAX_BACKUPS:
  678. raise AdminError("ROTATION_LIMIT", 3)
  679. prefix = f"pending-{request_digest[:16]}"
  680. result = _new_csr(root, args.common_name, prefix)
  681. record = {
  682. **result,
  683. "expected_current_certificate_digest": args.expected_current_certificate_digest,
  684. "request_digest": request_digest,
  685. "request_id": args.request_id,
  686. "status": "pending_server_approval",
  687. }
  688. _atomic_write(record_path, _json_bytes(record) + b"\n", 0o600)
  689. return record
  690. def _trusted_keys(bindings: list[str]) -> dict[str, bytes]:
  691. keys: dict[str, bytes] = {}
  692. for binding in bindings:
  693. key_id, separator, key_hex = binding.partition("=")
  694. if (
  695. not separator
  696. or not IDENTIFIER_RE.fullmatch(key_id)
  697. or not re.fullmatch(r"[0-9a-f]{64}", key_hex)
  698. ):
  699. raise AdminError("INVALID_INPUT")
  700. keys[key_id] = bytes.fromhex(key_hex)
  701. return keys
  702. STATUS_FIELDS = frozenset(
  703. {
  704. "certificate_sha256", "expires_at", "gateway_id", "generation",
  705. "issued_at", "key_id", "signature_algorithm", "status",
  706. "envelope_digest", "signature",
  707. }
  708. )
  709. def command_revoke_status(args: argparse.Namespace) -> dict[str, object]:
  710. root = _validated_root(args.root)
  711. local_marker = root / "state" / "REVOKED_STOP"
  712. status = "revoked" if local_marker.exists() else "not_revoked_locally"
  713. evidence_digest = None
  714. if args.server_status:
  715. evidence_path = _inside(root, args.server_status, must_exist=True)
  716. evidence = _read_json(evidence_path)
  717. if set(evidence) != STATUS_FIELDS:
  718. raise AdminError("STATUS_REJECTED", 3)
  719. unsigned = {
  720. key: evidence[key]
  721. for key in sorted(STATUS_FIELDS - {"envelope_digest", "signature"})
  722. }
  723. try:
  724. issued = datetime.fromisoformat(str(evidence["issued_at"]).replace("Z", "+00:00"))
  725. expires = datetime.fromisoformat(str(evidence["expires_at"]).replace("Z", "+00:00"))
  726. generation = int(evidence["generation"])
  727. except (TypeError, ValueError) as exc:
  728. raise AdminError("STATUS_REJECTED", 3) from exc
  729. now = datetime.now(UTC)
  730. if (
  731. not str(evidence["issued_at"]).endswith("Z")
  732. or not str(evidence["expires_at"]).endswith("Z")
  733. or issued.tzinfo is None
  734. or expires.tzinfo is None
  735. or issued.astimezone(UTC) > now
  736. or expires.astimezone(UTC) <= now
  737. or evidence["signature_algorithm"] != "Ed25519"
  738. or evidence["status"] not in {"active", "revoked"}
  739. or evidence["gateway_id"] != args.expected_gateway_id
  740. or generation != args.expected_generation
  741. or evidence["certificate_sha256"] != args.expected_certificate_digest
  742. or _protocol_digest(unsigned) != evidence["envelope_digest"]
  743. ):
  744. raise AdminError("STATUS_REJECTED", 3)
  745. public = _trusted_keys(args.trusted_key).get(str(evidence["key_id"]))
  746. if public is None:
  747. raise AdminError("STATUS_REJECTED", 3)
  748. try:
  749. ed25519.Ed25519PublicKey.from_public_bytes(public).verify(
  750. bytes.fromhex(str(evidence["signature"])), _protocol_bytes(unsigned)
  751. )
  752. except (InvalidSignature, ValueError) as exc:
  753. raise AdminError("STATUS_REJECTED", 3) from exc
  754. if evidence["status"] == "revoked":
  755. _atomic_write(local_marker, _json_bytes(evidence) + b"\n", 0o600)
  756. status = "revoked"
  757. evidence_digest = evidence["envelope_digest"]
  758. return {
  759. "evidence_digest": evidence_digest,
  760. "local_stop_required": status == "revoked",
  761. "status": status,
  762. }
  763. def _version_tuple(value: str) -> tuple[int, int, int, int]:
  764. if not VERSION_RE.fullmatch(value):
  765. raise AdminError("RELEASE_REJECTED", 3)
  766. values = tuple(int(item) for item in value.split("."))
  767. if any(item > 2_147_483_647 for item in values):
  768. raise AdminError("RELEASE_REJECTED", 3)
  769. return (*values, 0) # type: ignore[return-value]
  770. def _verify_release_manifest(
  771. manifest: dict[str, object], artifact_name_actual: str,
  772. artifact_digest_actual: str, trusted_keys: dict[str, bytes],
  773. *, current_version: str | None = None,
  774. ) -> dict[str, object]:
  775. if set(manifest) != MANIFEST_FIELDS:
  776. raise AdminError("RELEASE_REJECTED", 3)
  777. if not IDENTIFIER_RE.fullmatch(str(manifest["release_id"])):
  778. raise AdminError("RELEASE_REJECTED", 3)
  779. artifact_name = manifest["artifact_name"]
  780. if (
  781. not isinstance(artifact_name, str)
  782. or PurePosixPath(artifact_name).name != artifact_name
  783. or artifact_name != artifact_name_actual
  784. or artifact_name in {".", ".."}
  785. ):
  786. raise AdminError("RELEASE_REJECTED", 3)
  787. if (
  788. manifest["signature_algorithm"] != "Ed25519"
  789. or manifest["status"] not in {
  790. "offered", "accepted", "installed", "rolled_back"
  791. }
  792. ):
  793. raise AdminError("RELEASE_REJECTED", 3)
  794. version_tuple = _version_tuple(str(manifest["version"]))
  795. rollback_tuple = _version_tuple(str(manifest["rollback_version"]))
  796. if version_tuple <= rollback_tuple:
  797. raise AdminError("RELEASE_REJECTED", 3)
  798. if current_version is not None and (
  799. manifest["rollback_version"] != current_version
  800. or version_tuple <= _version_tuple(current_version)
  801. ):
  802. raise AdminError("RELEASE_REJECTED", 3)
  803. try:
  804. deadline = datetime.fromisoformat(str(manifest["deadline_at"]).replace("Z", "+00:00"))
  805. except ValueError as exc:
  806. raise AdminError("RELEASE_REJECTED", 3) from exc
  807. if deadline.tzinfo is None or deadline.astimezone(UTC) <= datetime.now(UTC):
  808. raise AdminError("RELEASE_REJECTED", 3)
  809. if not str(manifest["deadline_at"]).endswith("Z"):
  810. raise AdminError("RELEASE_REJECTED", 3)
  811. unsigned = {
  812. key: manifest[key]
  813. for key in sorted(MANIFEST_FIELDS - {"manifest_digest", "signature"})
  814. }
  815. if not SHA256_RE.fullmatch(str(manifest["manifest_digest"])) or _protocol_digest(unsigned) != manifest["manifest_digest"]:
  816. raise AdminError("RELEASE_REJECTED", 3)
  817. if not SHA256_RE.fullmatch(str(manifest["artifact_digest"])):
  818. raise AdminError("RELEASE_REJECTED", 3)
  819. if artifact_digest_actual != manifest["artifact_digest"]:
  820. raise AdminError("RELEASE_REJECTED", 3)
  821. public = trusted_keys.get(str(manifest["key_id"]))
  822. if public is None:
  823. raise AdminError("RELEASE_REJECTED", 3)
  824. try:
  825. ed25519.Ed25519PublicKey.from_public_bytes(public).verify(
  826. bytes.fromhex(str(manifest["signature"])), _protocol_bytes(unsigned)
  827. )
  828. except (ValueError, InvalidSignature) as exc:
  829. raise AdminError("RELEASE_REJECTED", 3) from exc
  830. return {
  831. "artifact_digest": artifact_digest_actual,
  832. "content_digest": manifest["manifest_digest"],
  833. "manifest": manifest,
  834. }
  835. def _bundle_record(manifest: dict[str, object]) -> dict[str, object]:
  836. unsigned = {
  837. "artifact_digest": manifest["artifact_digest"],
  838. "artifact_name": manifest["artifact_name"],
  839. "manifest_digest": manifest["manifest_digest"],
  840. "schema_version": 1,
  841. }
  842. return {**unsigned, "bundle_digest": _protocol_digest(unsigned)}
  843. def _verified_bundle(root: Path, manifest_digest: str) -> Path:
  844. if not SHA256_RE.fullmatch(manifest_digest):
  845. raise AdminError("RELEASE_REJECTED", 3)
  846. verified = root / "releases" / "verified"
  847. if verified.is_symlink() or not verified.is_dir():
  848. raise AdminError("PATH_POLICY")
  849. bundle = verified / manifest_digest
  850. if bundle.is_symlink() or not bundle.is_dir():
  851. raise AdminError("RELEASE_REJECTED", 3)
  852. return bundle
  853. def _verify_bundle(
  854. root: Path, manifest_digest: str, trusted_keys: dict[str, bytes]
  855. ) -> tuple[dict[str, object], dict[str, object], Path]:
  856. bundle = _verified_bundle(root, manifest_digest)
  857. manifest_path = _inside(root, bundle / "manifest.json", must_exist=True)
  858. manifest = _read_json(manifest_path)
  859. if manifest.get("manifest_digest") != manifest_digest:
  860. raise AdminError("RELEASE_REJECTED", 3)
  861. artifact_name = manifest.get("artifact_name")
  862. if not isinstance(artifact_name, str):
  863. raise AdminError("RELEASE_REJECTED", 3)
  864. artifact_path = _inside(root, bundle / artifact_name, must_exist=True)
  865. artifact_descriptor, artifact_metadata = _open_root_file(root, artifact_path)
  866. try:
  867. digest = hashlib.sha256()
  868. for chunk in iter(lambda: os.read(artifact_descriptor, 1_048_576), b""):
  869. digest.update(chunk)
  870. if os.fstat(artifact_descriptor) != artifact_metadata:
  871. raise AdminError("RELEASE_REJECTED", 3)
  872. finally:
  873. os.close(artifact_descriptor)
  874. _verify_release_manifest(manifest, artifact_name, digest.hexdigest(), trusted_keys)
  875. bundle_path = _inside(root, bundle / "bundle.json", must_exist=True)
  876. record = _read_json(bundle_path)
  877. expected = _bundle_record(manifest)
  878. if record != expected or set(bundle.iterdir()) != {
  879. manifest_path, artifact_path, bundle_path
  880. }:
  881. raise AdminError("RELEASE_REJECTED", 3)
  882. return manifest, record, artifact_path
  883. def command_release_check(args: argparse.Namespace) -> dict[str, object]:
  884. root = _validated_root(args.root)
  885. manifest_path = _inside(root, args.manifest)
  886. artifact_path = _inside(root, args.artifact)
  887. manifest_descriptor = -1
  888. artifact_descriptor = -1
  889. temporary: Path | None = None
  890. try:
  891. manifest_descriptor, manifest_before = _open_root_file(root, manifest_path)
  892. artifact_descriptor, artifact_before = _open_root_file(root, artifact_path)
  893. manifest = _read_json_descriptor(manifest_descriptor)
  894. if not _same_file_snapshot(os.fstat(manifest_descriptor), manifest_before):
  895. raise AdminError("RELEASE_REJECTED", 3)
  896. artifact_name = manifest.get("artifact_name")
  897. if (
  898. not isinstance(artifact_name, str)
  899. or not artifact_name
  900. or artifact_name in {".", ".."}
  901. or len(artifact_name.encode("utf-8")) > 255
  902. or PurePosixPath(artifact_name).name != artifact_name
  903. or Path(artifact_name).name != artifact_name
  904. or artifact_path.name != artifact_name
  905. ):
  906. raise AdminError("RELEASE_REJECTED", 3)
  907. verified_dir = _mkdir(root / "releases", "verified")
  908. temporary = Path(tempfile.mkdtemp(prefix=".bundle-", dir=verified_dir))
  909. os.chmod(temporary, 0o700)
  910. parent_owner = verified_dir.stat()
  911. if os.geteuid() == 0:
  912. os.chown(
  913. temporary, parent_owner.st_uid, parent_owner.st_gid,
  914. follow_symlinks=False,
  915. )
  916. elif (temporary.stat().st_uid, temporary.stat().st_gid) != (
  917. parent_owner.st_uid, parent_owner.st_gid
  918. ):
  919. raise AdminError("OWNER_POLICY")
  920. copied_artifact = temporary / artifact_name
  921. copied_digest = _copy_descriptor(
  922. artifact_descriptor, artifact_before, copied_artifact
  923. )
  924. keys = _trusted_keys(args.trusted_key)
  925. verified = _verify_release_manifest(
  926. manifest, artifact_path.name, copied_digest, keys,
  927. current_version=args.current_version,
  928. )
  929. replay_path = root / "state" / "release-replay.json"
  930. replay = _read_json(replay_path) if replay_path.exists() else {}
  931. content_digest = str(verified["content_digest"])
  932. prior = replay.get(str(manifest["release_id"]))
  933. if prior is not None and prior != content_digest:
  934. raise AdminError("RELEASE_REJECTED", 3)
  935. final_bundle = verified_dir / str(manifest["manifest_digest"])
  936. replayed = False
  937. if final_bundle.exists() or final_bundle.is_symlink():
  938. _verify_bundle(root, str(manifest["manifest_digest"]), keys)
  939. copied_artifact.unlink()
  940. temporary.rmdir()
  941. replayed = True
  942. else:
  943. _atomic_write(
  944. temporary / "manifest.json", _json_bytes(manifest) + b"\n", 0o600
  945. )
  946. _atomic_write(
  947. temporary / "bundle.json",
  948. _json_bytes(_bundle_record(manifest)) + b"\n", 0o600,
  949. )
  950. _fsync_directory(temporary)
  951. os.replace(temporary, final_bundle)
  952. _fsync_directory(verified_dir)
  953. replay[str(manifest["release_id"])] = content_digest
  954. _atomic_write(replay_path, _json_bytes(replay) + b"\n", 0o600)
  955. return {
  956. "artifact_digest": manifest["artifact_digest"],
  957. "manifest_digest": manifest["manifest_digest"],
  958. "replayed": replayed,
  959. "release_id": manifest["release_id"],
  960. "status": "verified",
  961. "verified_bundle_path": str(final_bundle),
  962. "version": manifest["version"],
  963. }
  964. finally:
  965. if manifest_descriptor >= 0:
  966. os.close(manifest_descriptor)
  967. if artifact_descriptor >= 0:
  968. os.close(artifact_descriptor)
  969. if temporary is not None and temporary.exists():
  970. if temporary.is_symlink() or not temporary.is_dir():
  971. raise AdminError("PATH_POLICY")
  972. for item in temporary.iterdir():
  973. if item.is_symlink() or not item.is_file():
  974. raise AdminError("PATH_POLICY")
  975. item.unlink()
  976. temporary.rmdir()
  977. _fsync_directory(temporary.parent)
  978. def command_rollback(args: argparse.Namespace) -> dict[str, object]:
  979. root = _validated_root(args.root)
  980. if not args.yes and (
  981. not sys.stdin.isatty()
  982. or input("Type ROLLBACK to continue: ").strip() != "ROLLBACK"
  983. ):
  984. raise AdminError("CONFIRMATION_REQUIRED", 4)
  985. if not SHA256_RE.fullmatch(args.expected_current_digest) or not SHA256_RE.fullmatch(args.target_digest):
  986. raise AdminError("INVALID_INPUT")
  987. keys = _trusted_keys(args.trusted_key)
  988. current_path = _inside(root, root / "releases" / "current.json", must_exist=True)
  989. current_pointer = _read_json(current_path)
  990. history_dir = _mkdir(root / "state", "rollback-history")
  991. request_digest = _protocol_digest(
  992. {
  993. "expected_current_digest": args.expected_current_digest,
  994. "target_digest": args.target_digest,
  995. }
  996. )
  997. history_path = history_dir / f"{request_digest}.json"
  998. intent_path = root / "state" / "rollback-intent.json"
  999. current, current_bundle, _ = _verify_bundle(
  1000. root, args.expected_current_digest, keys
  1001. )
  1002. target, target_bundle, _ = _verify_bundle(root, args.target_digest, keys)
  1003. expected_pointer = {
  1004. "artifact_name": current["artifact_name"],
  1005. "bundle_digest": current_bundle["bundle_digest"],
  1006. "manifest_digest": current["manifest_digest"],
  1007. "schema_version": 1,
  1008. "version": current["version"],
  1009. }
  1010. pointer = {
  1011. "artifact_name": target["artifact_name"],
  1012. "bundle_digest": target_bundle["bundle_digest"],
  1013. "manifest_digest": target["manifest_digest"],
  1014. "schema_version": 1,
  1015. "version": target["version"],
  1016. }
  1017. if (
  1018. not isinstance(target.get("version"), str)
  1019. or not isinstance(current.get("version"), str)
  1020. or _version_tuple(target["version"]) >= _version_tuple(current["version"])
  1021. or current.get("rollback_version") != target.get("version")
  1022. ):
  1023. raise AdminError("RELEASE_REJECTED", 3)
  1024. if history_path.exists():
  1025. history = _read_json(history_path)
  1026. if current_pointer != pointer:
  1027. raise AdminError("STATE_CONFLICT", 3)
  1028. if intent_path.exists():
  1029. expected_intent = {
  1030. "current_pointer": expected_pointer,
  1031. "request_digest": request_digest,
  1032. "target_pointer": pointer,
  1033. }
  1034. if _read_json(intent_path) != expected_intent:
  1035. raise AdminError("STATE_CONFLICT", 3)
  1036. intent_path.unlink()
  1037. _fsync_directory(intent_path.parent)
  1038. return {**history, "replayed": True}
  1039. if intent_path.exists():
  1040. intent = _read_json(intent_path)
  1041. if (
  1042. intent.get("request_digest") != request_digest
  1043. or intent.get("current_pointer") != expected_pointer
  1044. or intent.get("target_pointer") != pointer
  1045. ):
  1046. raise AdminError("STATE_CONFLICT", 3)
  1047. if current_pointer == pointer:
  1048. recovered = {
  1049. "current_manifest_digest": args.target_digest,
  1050. "previous_manifest_digest": args.expected_current_digest,
  1051. "replayed": True,
  1052. "status": "rolled_back",
  1053. "version": target.get("version"),
  1054. }
  1055. _atomic_write(history_path, _json_bytes(recovered) + b"\n", 0o600)
  1056. intent_path.unlink()
  1057. _fsync_directory(intent_path.parent)
  1058. return recovered
  1059. if current_pointer != expected_pointer:
  1060. raise AdminError("STATE_CONFLICT", 3)
  1061. backup_dir = _mkdir(root / "releases", "backups")
  1062. backup_path = backup_dir / f"current-{args.expected_current_digest}.json"
  1063. if not backup_path.exists():
  1064. _atomic_write(backup_path, _json_bytes(current_pointer) + b"\n", 0o600)
  1065. _bounded_backups(backup_dir, "current-")
  1066. intent = {
  1067. "current_pointer": expected_pointer,
  1068. "request_digest": request_digest,
  1069. "target_pointer": pointer,
  1070. }
  1071. if intent_path.exists() and _read_json(intent_path) != intent:
  1072. raise AdminError("STATE_CONFLICT", 3)
  1073. if not intent_path.exists():
  1074. _atomic_write(intent_path, _json_bytes(intent) + b"\n", 0o600)
  1075. if _read_json(current_path) != current_pointer:
  1076. raise AdminError("STATE_CONFLICT", 3)
  1077. _atomic_write(current_path, _json_bytes(pointer) + b"\n", 0o600)
  1078. result = {
  1079. "current_manifest_digest": args.target_digest,
  1080. "previous_manifest_digest": args.expected_current_digest,
  1081. "replayed": False,
  1082. "status": "rolled_back",
  1083. "version": target.get("version"),
  1084. }
  1085. _atomic_write(history_path, _json_bytes(result) + b"\n", 0o600)
  1086. intent_path.unlink()
  1087. _fsync_directory(intent_path.parent)
  1088. return result
  1089. def command_status(args: argparse.Namespace) -> dict[str, object]:
  1090. root = _validated_root(args.root)
  1091. queue_files = [
  1092. path for path in (root / "queue").glob("*.sqlite3")
  1093. if path.is_file() and not path.is_symlink()
  1094. ]
  1095. if len(queue_files) > 1:
  1096. raise AdminError("STATE_CONFLICT", 3)
  1097. queue_status: dict[str, object] = {
  1098. "artifact_cleanup": {}, "events": {}, "outcomes": {}, "releases": {},
  1099. "schema_version": None, "tasks": {},
  1100. }
  1101. active_leases = 0
  1102. if queue_files:
  1103. queue_path = _inside(root, queue_files[0], must_exist=True)
  1104. connection = sqlite3.connect(f"file:{queue_path}?mode=ro&immutable=1", uri=True)
  1105. try:
  1106. queue_status["schema_version"] = int(
  1107. connection.execute("PRAGMA user_version").fetchone()[0]
  1108. )
  1109. table_specs = {
  1110. "tasks": ("edge_tasks", "status"),
  1111. "events": ("edge_outbound_events", "status"),
  1112. "outcomes": ("edge_task_outcomes", "status"),
  1113. "artifact_cleanup": ("edge_local_artifacts", "cleanup_status"),
  1114. "releases": ("edge_release_state", "status"),
  1115. }
  1116. for label, (table, column) in table_specs.items():
  1117. rows = connection.execute(
  1118. f"SELECT {column},COUNT(*) FROM {table} GROUP BY {column}" # noqa: S608
  1119. ).fetchall()
  1120. queue_status[label] = {str(status): int(count) for status, count in rows}
  1121. active_leases = sum(
  1122. int(connection.execute(query).fetchone()[0])
  1123. for query in (
  1124. "SELECT COUNT(*) FROM edge_tasks WHERE status='leased'",
  1125. "SELECT COUNT(*) FROM edge_outbound_events WHERE status='sending'",
  1126. "SELECT COUNT(*) FROM edge_task_outcomes WHERE status='sending'",
  1127. "SELECT COUNT(*) FROM edge_local_artifacts WHERE cleanup_status='deleting'",
  1128. )
  1129. )
  1130. except sqlite3.DatabaseError as exc:
  1131. raise AdminError("QUEUE_STATUS_FAILED", 3) from exc
  1132. finally:
  1133. connection.close()
  1134. current = None
  1135. current_path = root / "releases" / "current.json"
  1136. if current_path.exists():
  1137. pointer = _read_json(current_path)
  1138. current = {
  1139. "manifest_digest": pointer.get("manifest_digest"),
  1140. "version": pointer.get("version"),
  1141. }
  1142. return {
  1143. "active_leases": active_leases,
  1144. "current_release": current,
  1145. "cursor_state": "reconciled_per_cycle_not_persisted",
  1146. "queue": queue_status,
  1147. "registration_present": (root / "registration" / "registration.json").is_file(),
  1148. "rollback_intent_present": (root / "state" / "rollback-intent.json").is_file(),
  1149. "status": "read_only",
  1150. }
  1151. def parser() -> argparse.ArgumentParser:
  1152. main = argparse.ArgumentParser(description=__doc__)
  1153. subcommands = main.add_subparsers(dest="command", required=True)
  1154. init = subcommands.add_parser("init")
  1155. init.add_argument("--root", required=True)
  1156. init.add_argument("--runtime-uid", required=True, type=int)
  1157. init.add_argument("--runtime-gid", required=True, type=int)
  1158. init.add_argument("--allow-chown", action="store_true")
  1159. init.set_defaults(handler=command_init)
  1160. csr = subcommands.add_parser("csr")
  1161. csr.add_argument("--root", required=True)
  1162. csr.add_argument("--common-name", required=True)
  1163. csr.set_defaults(handler=command_csr)
  1164. registration = subcommands.add_parser("register-manifest")
  1165. registration.add_argument("--root", required=True)
  1166. registration.add_argument("--certificate", required=True)
  1167. registration.add_argument("--ca-bundle", required=True)
  1168. registration.add_argument("--crl", required=True)
  1169. registration.add_argument("--gateway-id", required=True)
  1170. registration.add_argument("--environment", required=True)
  1171. registration.add_argument("--network-zone", required=True)
  1172. registration.add_argument("--policy-digest", required=True)
  1173. registration.add_argument("--control-origin", required=True)
  1174. registration.add_argument("--proxy-origin")
  1175. registration.set_defaults(handler=command_register_manifest)
  1176. rotate = subcommands.add_parser("rotate")
  1177. rotate.add_argument("--root", required=True)
  1178. rotate.add_argument("--common-name", required=True)
  1179. rotate.add_argument("--request-id", required=True)
  1180. rotate.add_argument("--expected-current-certificate-digest", required=True)
  1181. rotate.set_defaults(handler=command_rotate)
  1182. revoke = subcommands.add_parser("revoke-status")
  1183. revoke.add_argument("--root", required=True)
  1184. revoke.add_argument("--server-status")
  1185. revoke.add_argument("--trusted-key", action="append", default=[])
  1186. revoke.add_argument("--expected-gateway-id")
  1187. revoke.add_argument("--expected-generation", type=int)
  1188. revoke.add_argument("--expected-certificate-digest")
  1189. revoke.set_defaults(handler=command_revoke_status)
  1190. check = subcommands.add_parser("release-check")
  1191. check.add_argument("--root", required=True)
  1192. check.add_argument("--manifest", required=True)
  1193. check.add_argument("--artifact", required=True)
  1194. check.add_argument("--trusted-key", action="append", required=True)
  1195. check.add_argument("--current-version", required=True)
  1196. check.set_defaults(handler=command_release_check)
  1197. rollback = subcommands.add_parser("rollback")
  1198. rollback.add_argument("--root", required=True)
  1199. rollback.add_argument("--expected-current-digest", required=True)
  1200. rollback.add_argument("--target-digest", required=True)
  1201. rollback.add_argument("--trusted-key", action="append", required=True)
  1202. rollback.add_argument("--yes", action="store_true")
  1203. rollback.set_defaults(handler=command_rollback)
  1204. status_command = subcommands.add_parser("status")
  1205. status_command.add_argument("--root", required=True)
  1206. status_command.set_defaults(handler=command_status)
  1207. return main
  1208. def main() -> int:
  1209. try:
  1210. args = parser().parse_args()
  1211. if args.command in {"init", "status"}:
  1212. result = args.handler(args)
  1213. else:
  1214. root = _validated_root(args.root)
  1215. with _process_lock(root):
  1216. result = args.handler(args)
  1217. _output(result)
  1218. return 0
  1219. except AdminError as exc:
  1220. _output({"error": {"code": exc.code}, "status": "error"})
  1221. return exc.exit_code
  1222. except (OSError, ValueError, TypeError, MemoryError, RecursionError):
  1223. _output({"error": {"code": "OPERATION_FAILED"}, "status": "error"})
  1224. return 5
  1225. if __name__ == "__main__":
  1226. raise SystemExit(main())