"""Opt-in Linux ACL regression: PROD_FRONTEND_ACL_TESTS=1 (no CI changes).""" import json import os from pathlib import Path import shutil import socket import subprocess import sys import tempfile import time from types import SimpleNamespace import unittest import urllib.error import urllib.request from deployments.online.runtime import prod_frontend_acl as acl from scripts import serviceification_d3_engine as engine class PublicationHookTests(unittest.TestCase): def test_acl_failure_precedes_frontend_link_change(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() source = root / "source" source.mkdir() (source / "index.html").write_text("fixture") (source / "version.json").write_text(json.dumps({"version": "sha256:" + "b" * 64})) (root / "frontend-dist").symlink_to(source) context = SimpleNamespace(env_root=root, run_id="fixture", envv={"ENV_NAME": "multi-prod-d3-slot-3105"}) original = acl.ensure_frontend_acl def denied(release, target): self.assertEqual(root, release) self.assertEqual(source, (root / "frontend-dist").resolve()) self.assertTrue((target / "index.html").is_file()) raise ValueError("fixture access denied") acl.ensure_frontend_acl = denied try: with self.assertRaisesRegex(engine.DeployError, "before cutover"): engine.publish_readable_frontend_release(context, source, release_digest="sha256:" + "a" * 64) finally: acl.ensure_frontend_acl = original self.assertEqual(source, (root / "frontend-dist").resolve()) class LinuxACLTests(unittest.TestCase): @classmethod def setUpClass(cls): if sys.platform != "linux" or os.geteuid() != 0 or not shutil.which("nginx") or not shutil.which("sudo"): if os.environ.get("PROD_FRONTEND_ACL_TESTS") == "1": raise RuntimeError("ACL regression requires Linux/root/nginx/sudo/libacl and www-data UID 33") raise unittest.SkipTest("real ACL regression requires an isolated Linux root environment") def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name).resolve() self.root.chmod(0o755) self.release, self.frontend = self.make_release("staging-d3-release-fixture") self.paths = [self.release.parent.parent, self.release.parent, self.release, self.release / ".runtime", self.release / "run_hubspot_options_adapter.sh"] self.access = acl.AccessACL() self.before = {path: self.access.read(path) for path in self.paths} with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) self.port = sock.getsockname()[1] conf = self.root / "nginx.conf" conf.write_text(f"""user www-data; daemon off; pid {self.root}/nginx.pid; error_log {self.root}/error.log; events {{ worker_connections 32; }} http {{ server {{ listen 127.0.0.1:{self.port}; root {self.frontend}; location / {{ try_files $uri /index.html; }} }} }} """) log = (self.root / "startup.log").open("wb") self.addCleanup(log.close) self.process = subprocess.Popen([shutil.which("nginx"), "-p", str(self.root), "-c", str(conf)], stderr=log) def stop(): if self.process.poll() is None: self.process.terminate() self.process.wait(timeout=10) self.addCleanup(stop) deadline = time.monotonic() + 5 while time.monotonic() < deadline: users = subprocess.run(["ps", "-C", "nginx", "-o", "euid="], capture_output=True, text=True).stdout.split() if set(users) == {"0", "33"}: return time.sleep(0.05) self.fail((self.root / "startup.log").read_text()) def make_release(self, tag): release = self.root / "d3/releases" / tag release.mkdir(parents=True, mode=0o700) if not hasattr(self, "release"): release.parent.parent.chmod(0o700) release.parent.chmod(0o700) for name in (".runtime", "env", "logs", "run", "artifacts"): (release / name).mkdir(mode=0o755 if name == ".runtime" else 0o700) launcher = release / "run_hubspot_options_adapter.sh" launcher.write_text("fixture launcher") launcher.chmod(0o755) (release / "env/runtime.env").write_text("FIXTURE_ONLY=private") frontend = release / "frontend-releases" / ("d3-" + "a" * 64) frontend.mkdir(parents=True) (frontend / "index.html").write_text("fixture frontend") (frontend / "index.html").chmod(0o444) frontend.chmod(0o555) return release, frontend def http_status(self): try: with urllib.request.urlopen(f"http://127.0.0.1:{self.port}/index.html", timeout=5) as response: return response.status except urllib.error.HTTPError as response: response.close() return response.code def test_real_nginx_red_green_backup_retry_and_next_tag(self): self.assertEqual(500, self.http_status()) backup = acl.ensure_frontend_acl(self.release, self.frontend) self.assertEqual(200, self.http_status()) self.assertEqual(0o600, backup.stat().st_mode & 0o777) records = json.loads(backup.read_text())["paths"] self.assertEqual({str(p): text for p, text in self.before.items()}, {item["path"]: item["acl"] for item in records}) self.assertIsNone(acl.ensure_frontend_acl(self.release, self.frontend)) next_release, next_frontend = self.make_release("staging-d3-release-next") self.assertIsNotNone(acl.ensure_frontend_acl(next_release, next_frontend)) self.assertEqual(200, self.http_status()) self.assertEqual("FIXTURE_ONLY=private", (self.release / "env/runtime.env").read_text()) def test_privacy_failure_restores_exact_acl_and_retains_backup(self): (self.release / "env").chmod(0o755) with self.assertRaises(subprocess.CalledProcessError): acl.ensure_frontend_acl(self.release, self.frontend) self.assertEqual(self.before, {path: self.access.read(path) for path in self.paths}) self.assertEqual(1, len(list((self.release / ".runtime/frontend-acl-backups").glob("before-*.json")))) self.assertEqual(500, self.http_status()) def test_foreign_acl_rejected_without_mutation(self): parent = self.release.parent.parent self.access.write(parent, "user::rwx,user:34:--x,group::---,mask::--x,other::---") before = self.access.read(parent) with self.assertRaisesRegex(ValueError, "unreviewed"): acl.ensure_frontend_acl(self.release, self.frontend) self.assertEqual(before, self.access.read(parent)) self.assertFalse((self.release / ".runtime/frontend-acl-backups").exists()) if __name__ == "__main__": unittest.main()