mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-17 12:09:27 +00:00
Marius CiepluchandClaude Opus 4.8
6c7d6bb1d9
The path oracle only modelled classic parent traversal (a `canary` file in the parent of
${served_root}). A common real path-traversal class is the prefix-check bypass:
startswith(base) without a trailing separator, which the fix closes with base + os.sep;
it reaches a SIBLING whose path string-extends the served directory, not the parent.
PathOracle now also plants a sibling canary `public_leak` beside the served `public`
(same nonce, 0600), so a prefix-bypass demo leaks it on the vulnerable side and the fix
blocks it. ORACLE_GUIDE names the second target and tells the agent to pick the vector
its code shows. One shared nonce stays correct: a canary leaks only when the demo's own
request reaches it, and each demo targets one vector whose fix blocks it on the fixed side.
observe()/capture()/the differential/SideRecord are UNCHANGED. Local glance ceiling proof
still demonstrates (parent canary, 3/3 vuln, 0/3 fixed, control on both sides). Gate:
worker+control+verify suites 106 passed / 4 socket-skips; ruff clean; mypy clean on oracles.py.
Co-Authored-By: Claude Opus 4.8 <[email protected]>
348 lines
14 KiB
Python
348 lines
14 KiB
Python
"""Owned canaries. Only app captures enter these oracles, never attack output.
|
|
|
|
The attacker may supply CLI arguments through TARGET. Fixtures and app config
|
|
are mounted only in the app sandbox. Nonces are fresh per run and absent from
|
|
artefact env. A browser executor is trusted verifier code, never a demo script.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import secrets
|
|
import selectors
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
import tempfile
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
from typing import Protocol
|
|
from urllib.parse import quote
|
|
|
|
if __name__ != "__main__":
|
|
from . import _sandbox
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Canary:
|
|
root: Path
|
|
nonce: str
|
|
|
|
@classmethod
|
|
def fresh(cls, root: Path) -> Canary:
|
|
root.mkdir(mode=0o700)
|
|
return cls(root, secrets.token_hex(24))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ControlProbe:
|
|
"""A benign interaction a correctly functioning app must answer by returning
|
|
``token`` in its response body. It proves the demo's interface is wired and the
|
|
app can serve before any absence of the secret effect is trusted as a negative."""
|
|
|
|
method: str
|
|
path: str
|
|
token: str
|
|
body: str = ""
|
|
|
|
|
|
class Oracle(Protocol):
|
|
namespace_pid: int | None
|
|
placeholders: dict[str, str]
|
|
|
|
def prepare(self, canary: Canary) -> dict[str, str]: ...
|
|
def capture(self, output: str) -> None: ...
|
|
def observe(self) -> tuple[bool, str]: ...
|
|
def control(self) -> ControlProbe | None: ...
|
|
def close(self) -> None: ...
|
|
|
|
|
|
class OutputOracle:
|
|
namespace_pid: int | None = None
|
|
|
|
def prepare(self, canary: Canary) -> dict[str, str]:
|
|
self.canary = canary
|
|
self.seen = False
|
|
self.placeholders: dict[str, str] = {}
|
|
return {}
|
|
|
|
def capture(self, output: str) -> None:
|
|
self.seen |= self.canary.nonce in output
|
|
|
|
def observe(self) -> tuple[bool, str]:
|
|
return self.seen, "canary in verifier-captured app output" if self.seen else "no canary in app output"
|
|
|
|
def control(self) -> ControlProbe | None:
|
|
# No positive control by default. SQL/command/ssrf/xss controls are a
|
|
# documented follow-up; only the path oracle plants one for now.
|
|
return None
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
|
|
class PathOracle(OutputOracle):
|
|
control_name = "control"
|
|
|
|
def prepare(self, canary: Canary) -> dict[str, str]:
|
|
super().prepare(canary)
|
|
(canary.root / "public").mkdir()
|
|
(canary.root / "canary").write_text(canary.nonce)
|
|
(canary.root / "canary").chmod(0o600)
|
|
# Parent canary covers classic traversal; this sibling covers prefix checks
|
|
# using startswith without a separator.
|
|
(canary.root / "public_leak").write_text(canary.nonce)
|
|
(canary.root / "public_leak").chmod(0o600)
|
|
# A benign in-root control file, distinct from the out-of-root secret, so a
|
|
# correctly serving app returns it on BOTH revisions without any traversal.
|
|
# Drawn from os.urandom, not secrets.token_hex, so it never perturbs callers
|
|
# that monkeypatch the nonce source and count its draws.
|
|
self.control_token = os.urandom(24).hex()
|
|
control_file = canary.root / "public" / self.control_name
|
|
control_file.write_text(self.control_token)
|
|
control_file.chmod(0o600)
|
|
self.placeholders = {"served_root": "/fixture/public"}
|
|
return {"SERVED_ROOT": "/fixture/public"}
|
|
|
|
def control(self) -> ControlProbe | None:
|
|
return ControlProbe(method="GET", path="/" + self.control_name, token=self.control_token)
|
|
|
|
|
|
class SQLOracle(OutputOracle):
|
|
def prepare(self, canary: Canary) -> dict[str, str]:
|
|
super().prepare(canary)
|
|
with sqlite3.connect(canary.root / "fixture.db") as db:
|
|
db.execute("CREATE TABLE records (name TEXT, value TEXT, public INTEGER)")
|
|
db.executemany("INSERT INTO records VALUES (?, ?, ?)", [("ordinary", "public value", 1), ("secret", canary.nonce, 0)])
|
|
(canary.root / "fixture.db").chmod(0o600)
|
|
self.placeholders = {"database_path": "/fixture/fixture.db", "database_url": "sqlite:////fixture/fixture.db"}
|
|
return {"DATABASE": "/fixture/fixture.db"}
|
|
|
|
|
|
class CommandOracle(OutputOracle):
|
|
def prepare(self, canary: Canary) -> dict[str, str]:
|
|
super().prepare(canary)
|
|
# Marker is in app-only writable scratch, inaccessible in artefact sandbox.
|
|
self.marker = canary.root / "marker"
|
|
self.placeholders = {"marker_dir": "/scratch"}
|
|
return {}
|
|
|
|
def observe(self) -> tuple[bool, str]:
|
|
observed = self.marker.is_file() and not self.marker.is_symlink()
|
|
return observed, "injected command marker exists" if observed else "no command marker"
|
|
|
|
|
|
class SSRFOracle(OutputOracle):
|
|
"""Separate listener process and private loopback namespace; no external egress."""
|
|
|
|
task_boundary = False
|
|
process: subprocess.Popen[str] | None = None
|
|
|
|
def prepare(self, canary: Canary) -> dict[str, str]:
|
|
super().prepare(canary)
|
|
self.process = subprocess.Popen(
|
|
[
|
|
*(
|
|
[]
|
|
if self.task_boundary
|
|
else ["/usr/bin/unshare", *(["--user", "--map-root-user"] if _sandbox.isolation_mode() == "userns" else []), "--net"]
|
|
),
|
|
"/usr/bin/python3",
|
|
"-I",
|
|
str(Path(__file__).resolve()),
|
|
"--task-listener" if self.task_boundary else "--listener",
|
|
canary.nonce,
|
|
],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
env={"PATH": "/usr/bin:/bin"},
|
|
)
|
|
assert self.process.stdout is not None
|
|
with selectors.DefaultSelector() as selector:
|
|
selector.register(self.process.stdout, selectors.EVENT_READ)
|
|
if not selector.select(3):
|
|
raise OSError("listener namespace readiness timed out")
|
|
line = self.process.stdout.readline()
|
|
if not line:
|
|
raise OSError("private loopback listener unavailable")
|
|
self.namespace_pid, port = json.loads(line)
|
|
if self.task_boundary:
|
|
self.namespace_pid = None
|
|
self.placeholders = {"callback_url": f"http://127.0.0.1:{port}/{canary.nonce}"} # loopback: private oracle listener
|
|
return {"CALLBACK_URL": self.placeholders["callback_url"]}
|
|
|
|
def observe(self) -> tuple[bool, str]:
|
|
assert self.process is not None and self.process.stdout is not None
|
|
with selectors.DefaultSelector() as selector:
|
|
selector.register(self.process.stdout, selectors.EVENT_READ)
|
|
observed = bool(selector.select(0.2)) and self.process.stdout.readline().strip() == "observed"
|
|
return observed, "nonce request received" if observed else "no nonce request"
|
|
|
|
def close(self) -> None:
|
|
if self.process is not None:
|
|
self.process.terminate()
|
|
self.process.wait(timeout=3)
|
|
if self.process.stdout:
|
|
self.process.stdout.close()
|
|
|
|
|
|
class BrowserExecutor:
|
|
"""Run captured HTML with a verifier-owned alert canary.
|
|
|
|
Payloads call alert("ousast-xss"). Reflection/escaped HTML cannot set the fresh
|
|
DOM marker. A restrictive CSP disables external resources and connections;
|
|
Chromium's sandbox stays enabled for userns; root containers use the outer
|
|
boundary and drop the browser uid. Each capture uses a fresh profile.
|
|
Launch failures are unavailable execution, never a negative observation.
|
|
"""
|
|
|
|
def __init__(self, binary: str | None = None, *, timeout_seconds: float = 15, task_boundary: bool = False) -> None:
|
|
selected = binary or shutil.which("chromium") or shutil.which("chromium-browser")
|
|
self.timeout_seconds = timeout_seconds
|
|
if not selected:
|
|
raise OSError("headless chromium unavailable")
|
|
self.binary: str = selected
|
|
self.task_boundary = task_boundary
|
|
self.no_sandbox = task_boundary or _sandbox.isolation_mode() == "root-no-userns"
|
|
|
|
def __call__(self, document: str, nonce: str) -> bool:
|
|
marker = secrets.token_hex(24)
|
|
# Neither the marker nor its attribute appears in app output. The bootstrap
|
|
# source does appear in dump-dom, so inspect an actual DOM attribute only.
|
|
attribute = "data-proof-" + secrets.token_hex(12)
|
|
prelude = (
|
|
'<meta http-equiv="Content-Security-Policy" content="default-src \'none\'; '
|
|
"script-src 'unsafe-inline'; base-uri 'none'; form-action 'none'\">"
|
|
'<script>window.alert=function(value){if(value==="ousast-xss"||value==='
|
|
+ json.dumps(nonce)
|
|
+ "){document.documentElement.setAttribute("
|
|
+ json.dumps(attribute)
|
|
+ ","
|
|
+ json.dumps(marker)
|
|
+ ");}};</script>"
|
|
)
|
|
with tempfile.TemporaryDirectory(prefix="ousast-browser-") as profile:
|
|
# HTML can navigate even with CSP. A private network namespace, not
|
|
# browser flags, is the no-egress boundary. Expose runtime only.
|
|
if not self.task_boundary:
|
|
_sandbox.writable_directory(Path(profile))
|
|
namespaces = ["--unshare-pid", "--unshare-ipc", "--unshare-uts", "--unshare-net"] if self.no_sandbox else ["--unshare-all"]
|
|
sandbox = ["bwrap", *namespaces, "--die-with-parent", "--new-session", *_sandbox.capability_args()]
|
|
for path in ("/usr", "/bin", "/lib", "/lib64"):
|
|
if Path(path).exists():
|
|
sandbox += ["--ro-bind", path, path]
|
|
sandbox += [
|
|
"--dev",
|
|
"/dev",
|
|
"--proc",
|
|
"/proc",
|
|
"--tmpfs",
|
|
"/tmp",
|
|
"--bind",
|
|
profile,
|
|
"/profile",
|
|
"--setenv",
|
|
"HOME",
|
|
"/profile",
|
|
"--",
|
|
]
|
|
if self.task_boundary:
|
|
sandbox = []
|
|
try:
|
|
done = subprocess.run(
|
|
[
|
|
*sandbox,
|
|
*(_sandbox.drop_privileges() if self.no_sandbox and not self.task_boundary else []),
|
|
self.binary,
|
|
*(["--no-sandbox"] if self.no_sandbox else []),
|
|
"--headless",
|
|
"--dump-dom",
|
|
"--disable-gpu",
|
|
"--disable-background-networking",
|
|
"--disable-extensions",
|
|
"--no-first-run",
|
|
"--no-default-browser-check",
|
|
"--disable-sync",
|
|
"--user-data-dir=" + (profile if self.task_boundary else "/profile"),
|
|
"--virtual-time-budget=1000",
|
|
"data:text/html;charset=utf-8," + quote(prelude + document),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=self.timeout_seconds,
|
|
env={"PATH": "/usr/bin:/bin", "HOME": profile},
|
|
)
|
|
except subprocess.SubprocessError as exc:
|
|
raise OSError("headless chromium failed") from exc
|
|
if done.returncode or "<html" not in done.stdout.lower():
|
|
raise OSError("headless chromium did not produce a document: " + done.stderr[-500:])
|
|
|
|
class Marker(HTMLParser):
|
|
seen = False
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
if tag == "html" and (attribute, marker) in attrs:
|
|
self.seen = True
|
|
|
|
parser = Marker()
|
|
parser.feed(done.stdout)
|
|
return parser.seen
|
|
|
|
|
|
Browser = Callable[[str, str], bool]
|
|
|
|
|
|
class XSSOracle(OutputOracle):
|
|
def __init__(self, executor: Browser) -> None:
|
|
self.executor = executor
|
|
|
|
def capture(self, output: str) -> None:
|
|
self.seen |= self.executor(output, self.canary.nonce)
|
|
|
|
|
|
def oracle_for(family: str, browser: Browser | None = None) -> Oracle | None:
|
|
factories = {"path": PathOracle, "sql": SQLOracle, "command": CommandOracle, "ssrf": SSRFOracle}
|
|
if family == "xss":
|
|
return XSSOracle(browser) if browser else None
|
|
factory = factories.get(family)
|
|
return factory() if factory else None
|
|
|
|
|
|
def _listen(nonce: str, *, task_boundary: bool = False) -> None:
|
|
import fcntl
|
|
import socket
|
|
import struct
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
# unshare created only loopback; bring it up without iproute2 or any egress.
|
|
if not task_boundary:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
fcntl.ioctl(sock, 0x8914, struct.pack("16sH14s", b"lo", 0x49, b""))
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None:
|
|
if self.path.split("?", 1)[0] == "/" + nonce:
|
|
print("observed", flush=True)
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(b"ok")
|
|
|
|
do_POST = do_PUT = do_PATCH = do_DELETE = do_HEAD = do_OPTIONS = do_TRACE = do_CONNECT = do_GET
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
pass
|
|
|
|
with HTTPServer(("127.0.0.1", 0), Handler) as server:
|
|
print(json.dumps([os.getpid(), server.server_port]), flush=True)
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
_listen(sys.argv[2], task_boundary=sys.argv[1] == "--task-listener")
|