mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-10-08 16:02:20 +02:00
fix(security): harden Python service boundaries
This commit is contained in:
Alexandre Teixeira
@@ -123,3 +123,37 @@ async def test_update_event_can_add_reminder_after_creation():
|
||||
assert note is not None, "update_event should create the reminder note"
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
async def test_reminder_and_duration_parsers_stay_linear_on_digit_and_space_floods():
|
||||
"""CodeQL py/polynomial-redos: `(\\d+)\\s*unit` rescanned a digit run from
|
||||
every offset and `alarm\\s*:?\\s*\\d+` split one whitespace run two ways.
|
||||
Tool arguments come from model output, so keep them O(n)."""
|
||||
import time
|
||||
from src.tool_implementations import do_manage_calendar
|
||||
|
||||
owner = "tester-" + uuid.uuid4().hex[:6]
|
||||
started = time.perf_counter()
|
||||
res = await do_manage_calendar(json.dumps({
|
||||
"action": "create_event",
|
||||
"summary": "Flood",
|
||||
"dtstart": "2030-04-01T10:00:00",
|
||||
"reminder_minutes": "0" * 40_000 + "x",
|
||||
"duration": "0" * 40_000 + "x",
|
||||
}), owner=owner)
|
||||
assert time.perf_counter() - started < 2.0
|
||||
assert res.get("exit_code") == 0, res
|
||||
assert "reminder set" not in res.get("response", ""), res
|
||||
|
||||
# A reminder is set, so the "is the description only a reminder?" check runs.
|
||||
started = time.perf_counter()
|
||||
res = await do_manage_calendar(json.dumps({
|
||||
"action": "create_event",
|
||||
"summary": "Flood 2",
|
||||
"dtstart": "2030-04-02T10:00:00",
|
||||
"description": "alarm" + "\t" * 40_000,
|
||||
"reminder_minutes": 5,
|
||||
}), owner=owner)
|
||||
assert time.perf_counter() - started < 2.0
|
||||
assert res.get("exit_code") == 0, res
|
||||
assert "reminder 5 min before" in res.get("response", ""), res
|
||||
|
||||
@@ -0,0 +1,522 @@
|
||||
"""Final closure tests for the residual Python CodeQL lane (PR #6503).
|
||||
|
||||
1. MCP attachment extraction: no traversal, per owner/account directories.
|
||||
2. Mail DNS rebinding: one resolution, connect only to the checked address,
|
||||
TLS SNI/verification still against the requested hostname.
|
||||
3. /accounts/test runs its blocking mail I/O off the event loop.
|
||||
4. /search/web works under the production CSP and payloads cannot execute.
|
||||
5. Outbound mail address policy matrix.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
import urllib.parse
|
||||
from email.mime.application import MIMEApplication
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# -- 1. MCP attachment extraction -------------------------------------------
|
||||
|
||||
def _message_with_attachment(body: bytes) -> bytes:
|
||||
msg = MIMEMultipart()
|
||||
part = MIMEApplication(body, Name="invoice.pdf")
|
||||
part["Content-Disposition"] = 'attachment; filename="invoice.pdf"'
|
||||
msg.attach(part)
|
||||
return msg.as_bytes()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mcp_mail(tmp_path, monkeypatch):
|
||||
import mcp_servers.email_server as es
|
||||
|
||||
root = tmp_path / "mail-attachments"
|
||||
root.mkdir()
|
||||
monkeypatch.setattr(es, "MAIL_ATTACHMENTS_DIR", str(root))
|
||||
monkeypatch.setattr(es, "_fixture_attachment_source", lambda *a, **k: None)
|
||||
monkeypatch.setattr(es, "_load_config", lambda account=None: {"account_id": account})
|
||||
box = {"raw": _message_with_attachment(b"%PDF one")}
|
||||
|
||||
class _Conn:
|
||||
def select(self, *a, **k):
|
||||
return ("OK", [b"1"])
|
||||
|
||||
def uid(self, *a):
|
||||
return ("OK", [(b"42 (BODY[])", box["raw"])])
|
||||
|
||||
def logout(self):
|
||||
pass
|
||||
|
||||
monkeypatch.setattr(es, "_imap_connect", lambda account=None: _Conn())
|
||||
return es, root.resolve(), box
|
||||
|
||||
|
||||
@pytest.mark.parametrize("folder,uid", [
|
||||
("/tmp/odysseus-escape", "42"), # absolute mailbox name replaced the root
|
||||
("../../escape", "42"), # relative traversal
|
||||
("INBOX/../../../escape", "42"), # hierarchy delimiter + traversal
|
||||
("..", ".."),
|
||||
("INBOX", "../../42"),
|
||||
("INBOX/Receipts", "42"), # legitimate hierarchical folder
|
||||
])
|
||||
def test_mcp_download_attachment_stays_in_one_root_segment(mcp_mail, folder, uid):
|
||||
es, root, _box = mcp_mail
|
||||
result = es._download_attachment(uid, 0, folder)
|
||||
path = Path(result["path"]).resolve()
|
||||
assert path.parent.parent == root, path
|
||||
assert path.read_bytes() == b"%PDF one"
|
||||
|
||||
|
||||
def test_mcp_download_attachment_isolates_owners_and_accounts(mcp_mail):
|
||||
es, _root, box = mcp_mail
|
||||
paths = {}
|
||||
for owner, account, body in (
|
||||
("alice", "acct-a", b"%PDF alice"),
|
||||
("bob", "acct-a", b"%PDF bob"),
|
||||
("alice", "acct-b", b"%PDF alice second mailbox"),
|
||||
):
|
||||
token = es._CURRENT_OWNER.set(owner)
|
||||
try:
|
||||
box["raw"] = _message_with_attachment(body)
|
||||
paths[(owner, account)] = (es._download_attachment("42", 0, "INBOX", account=account)["path"], body)
|
||||
finally:
|
||||
es._CURRENT_OWNER.reset(token)
|
||||
assert len({p for p, _ in paths.values()}) == 3
|
||||
for path, body in paths.values():
|
||||
assert Path(path).read_bytes() == body
|
||||
|
||||
|
||||
def test_attachment_scope_dir_rejects_symlinked_scope(tmp_path):
|
||||
from src.mail_attachment_paths import attachment_scope_dir
|
||||
|
||||
root = tmp_path / "root"
|
||||
root.mkdir()
|
||||
target = attachment_scope_dir(root, "INBOX", "1", owner="o", account_id="a")
|
||||
outside = tmp_path / "outside"
|
||||
outside.mkdir()
|
||||
target.symlink_to(outside, target_is_directory=True)
|
||||
with pytest.raises(ValueError):
|
||||
attachment_scope_dir(root, "INBOX", "1", owner="o", account_id="a")
|
||||
|
||||
|
||||
def test_attachment_scope_dir_normalises_uid_type(tmp_path):
|
||||
from src.mail_attachment_paths import attachment_scope_dir
|
||||
|
||||
assert attachment_scope_dir(tmp_path, "INBOX", 42, owner="o", account_id=None) == \
|
||||
attachment_scope_dir(tmp_path, "INBOX", "42", owner="o", account_id="")
|
||||
|
||||
|
||||
# -- 2. DNS rebinding: resolve once, connect to the checked address ------------
|
||||
|
||||
class _LineServer:
|
||||
"""Tiny loopback server speaking just enough IMAP or SMTP for a handshake."""
|
||||
|
||||
def __init__(self, protocol):
|
||||
self.protocol = protocol
|
||||
self.sock = socket.socket()
|
||||
self.sock.bind(("127.0.0.1", 0))
|
||||
self.sock.listen(4)
|
||||
self.port = self.sock.getsockname()[1]
|
||||
self.thread = threading.Thread(target=self._serve, daemon=True)
|
||||
self.thread.start()
|
||||
|
||||
def _serve(self):
|
||||
try:
|
||||
conn, _ = self.sock.accept()
|
||||
except OSError:
|
||||
return
|
||||
with conn, conn.makefile("rwb") as io:
|
||||
io.write(b"* OK ready\r\n" if self.protocol == "imap" else b"220 fake ESMTP\r\n")
|
||||
io.flush()
|
||||
for raw in io:
|
||||
line = raw.decode().strip()
|
||||
if self.protocol == "imap":
|
||||
tag, _, command = line.partition(" ")
|
||||
if command.upper().startswith("CAPABILITY"):
|
||||
io.write(f"* CAPABILITY IMAP4rev1\r\n{tag} OK done\r\n".encode())
|
||||
else:
|
||||
io.write(f"* BYE\r\n{tag} OK bye\r\n".encode())
|
||||
io.flush()
|
||||
return
|
||||
elif line.upper().startswith(("EHLO", "HELO")):
|
||||
io.write(b"250 fake\r\n")
|
||||
else:
|
||||
io.write(b"221 bye\r\n")
|
||||
io.flush()
|
||||
return
|
||||
io.flush()
|
||||
|
||||
def close(self):
|
||||
self.sock.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rebinding_dns(monkeypatch):
|
||||
"""`rebind.test` answers loopback once, then the metadata address."""
|
||||
real = socket.getaddrinfo
|
||||
lookups = []
|
||||
|
||||
def _resolve(host, port, *args, **kwargs):
|
||||
if host == "rebind.test":
|
||||
lookups.append(host)
|
||||
ip = "127.0.0.1" if len(lookups) == 1 else "169.254.169.254"
|
||||
return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, port))]
|
||||
return real(host, port, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(socket, "getaddrinfo", _resolve)
|
||||
return lookups
|
||||
|
||||
|
||||
class _RecordingTLS:
|
||||
"""Stands in for an SSLContext: records the SNI name, keeps the socket plain."""
|
||||
|
||||
def __init__(self):
|
||||
self.server_hostname = None
|
||||
|
||||
def wrap_socket(self, sock, server_hostname=None, **_kwargs):
|
||||
self.server_hostname = server_hostname
|
||||
return sock
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tls", [False, True])
|
||||
def test_imap_connects_to_the_checked_address_only(rebinding_dns, tls):
|
||||
from routes.email.email_helpers import _PolicyIMAP4, _PolicyIMAP4_SSL
|
||||
|
||||
server = _LineServer("imap")
|
||||
try:
|
||||
if tls:
|
||||
context = _RecordingTLS()
|
||||
conn = _PolicyIMAP4_SSL("rebind.test", server.port, block_private=False, timeout=5, ssl_context=context)
|
||||
assert context.server_hostname == "rebind.test"
|
||||
else:
|
||||
conn = _PolicyIMAP4("rebind.test", server.port, block_private=False, timeout=5)
|
||||
assert conn.sock.getpeername()[0] == "127.0.0.1"
|
||||
conn.logout()
|
||||
finally:
|
||||
server.close()
|
||||
assert rebinding_dns == ["rebind.test"] # a second lookup would have answered 169.254.169.254
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tls", [False, True])
|
||||
def test_smtp_connects_to_the_checked_address_only(rebinding_dns, tls):
|
||||
from routes.email.email_helpers import _PolicySMTP, _PolicySMTP_SSL
|
||||
|
||||
server = _LineServer("smtp")
|
||||
try:
|
||||
if tls:
|
||||
context = _RecordingTLS()
|
||||
smtp = _PolicySMTP_SSL("rebind.test", server.port, block_private=False, timeout=5, context=context)
|
||||
assert context.server_hostname == "rebind.test"
|
||||
else:
|
||||
smtp = _PolicySMTP("rebind.test", server.port, block_private=False, timeout=5)
|
||||
assert smtp.sock.getpeername()[0] == "127.0.0.1"
|
||||
assert smtp.ehlo()[0] == 250
|
||||
smtp.quit()
|
||||
finally:
|
||||
server.close()
|
||||
assert rebinding_dns == ["rebind.test"]
|
||||
|
||||
|
||||
def test_any_denied_answer_in_a_mixed_resolution_blocks_the_connection():
|
||||
from src.url_safety import OutboundAddressBlocked, connect_outbound_tcp
|
||||
|
||||
def _mixed(host, port, *args):
|
||||
return [
|
||||
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", port)),
|
||||
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("169.254.169.254", port)),
|
||||
]
|
||||
|
||||
with pytest.raises(OutboundAddressBlocked):
|
||||
connect_outbound_tcp("mixed.test", 993, block_private=False, resolver=_mixed)
|
||||
|
||||
|
||||
# -- 5. Address/principal policy matrix ----------------------------------------
|
||||
|
||||
_ALWAYS_DENIED = [
|
||||
"169.254.169.254", "fe80::1", "0.0.0.0", "::", "224.0.0.1", "240.0.0.1",
|
||||
"::ffff:169.254.169.254", "64:ff9b::a9fe:a9fe", "64:ff9b::a00:5",
|
||||
"::1", # IPv6 loopback sits in ::/8, which the existing policy treats as reserved
|
||||
]
|
||||
_PRIVATE = ["127.0.0.1", "10.0.0.5", "172.16.0.1", "192.168.1.10", "100.64.0.1", "fd00::1"]
|
||||
_PUBLIC = ["93.184.216.34", "2606:2800:220:1::1"]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_sockets(monkeypatch):
|
||||
import src.url_safety as url_safety
|
||||
|
||||
connected = []
|
||||
|
||||
class _Sock:
|
||||
def __init__(self, *args):
|
||||
pass
|
||||
|
||||
def settimeout(self, timeout):
|
||||
pass
|
||||
|
||||
def bind(self, address):
|
||||
pass
|
||||
|
||||
def connect(self, sockaddr):
|
||||
connected.append(sockaddr[0])
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
monkeypatch.setattr(url_safety.socket, "socket", _Sock)
|
||||
return connected
|
||||
|
||||
|
||||
def _answer(ip):
|
||||
family = socket.AF_INET6 if ":" in ip else socket.AF_INET
|
||||
return lambda host, port, *args: [(family, socket.SOCK_STREAM, 6, "", (ip, port))]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("block_private", [False, True])
|
||||
@pytest.mark.parametrize("ip", _ALWAYS_DENIED)
|
||||
def test_metadata_and_special_ranges_are_always_denied(fake_sockets, ip, block_private):
|
||||
from src.url_safety import OutboundAddressBlocked, connect_outbound_tcp
|
||||
|
||||
with pytest.raises(OutboundAddressBlocked):
|
||||
connect_outbound_tcp("h", 993, block_private=block_private, resolver=_answer(ip))
|
||||
assert fake_sockets == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ip", _PRIVATE)
|
||||
def test_private_ranges_follow_the_principal_policy(fake_sockets, ip):
|
||||
from src.url_safety import OutboundAddressBlocked, connect_outbound_tcp
|
||||
|
||||
connect_outbound_tcp("h", 993, block_private=False, resolver=_answer(ip))
|
||||
assert fake_sockets == [ip]
|
||||
with pytest.raises(OutboundAddressBlocked):
|
||||
connect_outbound_tcp("h", 993, block_private=True, resolver=_answer(ip))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ip", _PUBLIC)
|
||||
def test_public_mail_servers_are_always_allowed(fake_sockets, ip):
|
||||
from src.url_safety import connect_outbound_tcp
|
||||
|
||||
connect_outbound_tcp("h", 993, block_private=True, resolver=_answer(ip))
|
||||
assert fake_sockets == [ip]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("trusted,env,owner,blocked", [
|
||||
(False, {}, "bob", True), # multi-user, non-admin: denied
|
||||
(False, {}, "", True), # multi-user, no principal: denied
|
||||
(True, {}, "admin", False), # admin (or single-user mode)
|
||||
(False, {"EMAIL_ALLOW_PRIVATE_IPS": "true"}, "bob", False), # operator opt-in for LAN mail
|
||||
(True, {"EMAIL_BLOCK_PRIVATE_IPS": "true"}, "admin", True), # operator lockdown wins
|
||||
(False, {"EMAIL_ALLOW_PRIVATE_IPS": "true", "EMAIL_BLOCK_PRIVATE_IPS": "true"}, "bob", True),
|
||||
])
|
||||
def test_private_mail_destination_principal_matrix(monkeypatch, trusted, env, owner, blocked):
|
||||
import src.tool_security as tool_security
|
||||
from routes.email.email_helpers import _mail_private_blocked
|
||||
|
||||
for key in ("EMAIL_ALLOW_PRIVATE_IPS", "EMAIL_BLOCK_PRIVATE_IPS"):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
for key, value in env.items():
|
||||
monkeypatch.setenv(key, value)
|
||||
monkeypatch.setattr(tool_security, "owner_is_admin_or_single_user", lambda o: trusted)
|
||||
assert _mail_private_blocked(owner) is blocked
|
||||
|
||||
|
||||
def test_single_user_mode_allows_lan_mail_servers(monkeypatch):
|
||||
from routes.email.email_helpers import _mail_private_blocked
|
||||
|
||||
for key in ("EMAIL_ALLOW_PRIVATE_IPS", "EMAIL_BLOCK_PRIVATE_IPS"):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
monkeypatch.setenv("AUTH_ENABLED", "false")
|
||||
assert _mail_private_blocked("") is False
|
||||
|
||||
|
||||
def test_non_admin_cannot_probe_loopback_through_the_connection_test(monkeypatch):
|
||||
import routes.email_routes as email_routes
|
||||
import src.tool_security as tool_security
|
||||
|
||||
for key in ("EMAIL_ALLOW_PRIVATE_IPS", "EMAIL_BLOCK_PRIVATE_IPS"):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
monkeypatch.setattr(tool_security, "owner_is_admin_or_single_user", lambda o: False)
|
||||
endpoint = _accounts_test_endpoint(email_routes)
|
||||
|
||||
class _Request:
|
||||
async def json(self):
|
||||
return {"imap_host": "127.0.0.1", "imap_port": 6379, "imap_user": "u", "imap_password": "p",
|
||||
"smtp_host": "10.0.0.5", "smtp_port": 25}
|
||||
|
||||
result = asyncio.run(endpoint(req=_Request(), owner="bob"))
|
||||
assert result["imap"]["error"] == "IMAP server address is not allowed by this server's network policy"
|
||||
assert result["smtp"]["error"] == "SMTP server address is not allowed by this server's network policy"
|
||||
|
||||
|
||||
# -- 3. /accounts/test blocking I/O stays off the event loop -----------------------
|
||||
|
||||
def _accounts_test_endpoint(email_routes):
|
||||
router = email_routes.setup_email_routes()
|
||||
return next(
|
||||
r.endpoint for r in router.routes
|
||||
if r.path == "/api/email/accounts/test" and "POST" in getattr(r, "methods", set())
|
||||
)
|
||||
|
||||
|
||||
def test_connection_test_does_not_block_the_event_loop(monkeypatch):
|
||||
import routes.email_routes as email_routes
|
||||
|
||||
def _slow_open(host, port, **kwargs):
|
||||
time.sleep(1.0) # a mail server that never answers within the timeout
|
||||
raise socket.timeout("timed out")
|
||||
|
||||
monkeypatch.setattr(email_routes, "_open_imap_connection", _slow_open)
|
||||
endpoint = _accounts_test_endpoint(email_routes)
|
||||
|
||||
class _Request:
|
||||
async def json(self):
|
||||
return {"imap_host": "imap.example.com", "imap_port": 993, "imap_user": "u", "imap_password": "p"}
|
||||
|
||||
async def _main():
|
||||
gaps = []
|
||||
done = asyncio.Event()
|
||||
|
||||
async def _heartbeat():
|
||||
last = time.perf_counter()
|
||||
while not done.is_set():
|
||||
await asyncio.sleep(0.02)
|
||||
now = time.perf_counter()
|
||||
gaps.append(now - last)
|
||||
last = now
|
||||
|
||||
beat = asyncio.create_task(_heartbeat())
|
||||
started = time.perf_counter()
|
||||
results = await asyncio.gather(*(endpoint(req=_Request(), owner="alice") for _ in range(3)))
|
||||
elapsed = time.perf_counter() - started
|
||||
done.set()
|
||||
await beat
|
||||
return results, max(gaps), elapsed
|
||||
|
||||
results, worst_gap, elapsed = asyncio.run(_main())
|
||||
assert worst_gap < 0.5, f"event loop stalled for {worst_gap:.2f}s"
|
||||
assert elapsed < 2.5 # three 1s probes overlapped instead of serialising on the loop
|
||||
for result in results:
|
||||
assert result == {"ok": False, "imap": {"ok": False, "error": "IMAP connection timed out"}, "smtp": None}
|
||||
|
||||
|
||||
def test_connection_test_keeps_its_authentication_dependency():
|
||||
import routes.email_routes as email_routes
|
||||
from routes.email.email_routes import require_user
|
||||
|
||||
router = email_routes.setup_email_routes()
|
||||
route = next(
|
||||
r for r in router.routes
|
||||
if r.path == "/api/email/accounts/test" and "POST" in getattr(r, "methods", set())
|
||||
)
|
||||
assert any(dep.call is require_user for dep in route.dependant.dependencies)
|
||||
|
||||
|
||||
# -- 4. /search/web under the production CSP ---------------------------------------
|
||||
|
||||
_PAYLOAD = (
|
||||
"</script><script>document.title='PWNED'</script>"
|
||||
"<img src=x onerror=\"document.body.dataset.pwned='1'\">"
|
||||
)
|
||||
_SOURCES = [
|
||||
{"title": "Example result", "url": "https://example.com/a", "snippet": "first"},
|
||||
{"title": "Script link", "url": "javascript:document.body.dataset.pwned='2'", "snippet": "second"},
|
||||
]
|
||||
|
||||
|
||||
def _csp_app():
|
||||
from fastapi import FastAPI
|
||||
from fastapi.responses import HTMLResponse
|
||||
from core.middleware import SecurityHeadersMiddleware
|
||||
from routes.search.search_routes import setup_search_routes
|
||||
|
||||
app = FastAPI()
|
||||
app.add_middleware(SecurityHeadersMiddleware)
|
||||
|
||||
@app.post("/api/search")
|
||||
async def _stub_search():
|
||||
return {"sources": _SOURCES}
|
||||
|
||||
@app.get("/csp-control")
|
||||
async def _control():
|
||||
return HTMLResponse("<html><body><script>document.body.dataset.ran='1'</script></body></html>")
|
||||
|
||||
app.include_router(setup_search_routes(SimpleNamespace()))
|
||||
return app
|
||||
|
||||
|
||||
@pytest.mark.parametrize("q", ["weather in lisbon", _PAYLOAD])
|
||||
def test_search_page_script_carries_the_response_csp_nonce(q):
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
response = TestClient(_csp_app()).get("/search/web", params={"q": q})
|
||||
csp = response.headers["content-security-policy"]
|
||||
nonce = re.search(r"'nonce-([^']+)'", csp).group(1)
|
||||
assert "unsafe-inline" not in csp.split("script-src", 1)[1].split(";", 1)[0]
|
||||
scripts = re.findall(r"<script\b[^>]*>", response.text, re.I)
|
||||
assert scripts == [f'<script nonce="{nonce}">']
|
||||
markup = response.text.split("<script", 1)[0]
|
||||
assert "<img" not in markup.lower() # the payload is only ever present HTML-escaped
|
||||
|
||||
|
||||
def _chrome():
|
||||
for name in ("google-chrome", "google-chrome-stable", "chromium", "chromium-browser"):
|
||||
path = shutil.which(name)
|
||||
if path:
|
||||
return path
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def live_csp_server():
|
||||
uvicorn = pytest.importorskip("uvicorn")
|
||||
with socket.socket() as probe:
|
||||
probe.bind(("127.0.0.1", 0))
|
||||
port = probe.getsockname()[1]
|
||||
server = uvicorn.Server(uvicorn.Config(_csp_app(), host="127.0.0.1", port=port, log_level="error"))
|
||||
thread = threading.Thread(target=server.run, daemon=True)
|
||||
thread.start()
|
||||
deadline = time.time() + 10
|
||||
while not server.started and time.time() < deadline:
|
||||
time.sleep(0.05)
|
||||
yield f"http://127.0.0.1:{port}"
|
||||
server.should_exit = True
|
||||
thread.join(timeout=10)
|
||||
|
||||
|
||||
def _render(chrome, url, profile):
|
||||
out = subprocess.run(
|
||||
[chrome, "--headless=new", "--disable-gpu", "--no-sandbox", "--no-first-run",
|
||||
f"--user-data-dir={profile}", "--virtual-time-budget=5000", "--dump-dom", url],
|
||||
capture_output=True, text=True, timeout=90,
|
||||
)
|
||||
return out.stdout
|
||||
|
||||
|
||||
@pytest.mark.skipif(_chrome() is None, reason="needs a headless Chrome/Chromium")
|
||||
def test_search_page_runs_under_production_csp_and_payload_does_not(live_csp_server, tmp_path):
|
||||
chrome = _chrome()
|
||||
# Control: the production CSP really blocks nonce-less inline script.
|
||||
control = _render(chrome, f"{live_csp_server}/csp-control", tmp_path / "p0")
|
||||
assert "<body>" in control and "data-ran" not in control
|
||||
|
||||
normal = _render(chrome, f"{live_csp_server}/search/web?q=weather", tmp_path / "p1")
|
||||
assert "2 results" in normal # the page's script executed
|
||||
assert 'href="https://example.com/a"' in normal
|
||||
assert "javascript:" not in normal.split('id="results"', 1)[1] # non-http result URL not linked
|
||||
|
||||
hostile = _render(
|
||||
chrome, f"{live_csp_server}/search/web?q={urllib.parse.quote(_PAYLOAD)}", tmp_path / "p2",
|
||||
)
|
||||
assert "2 results" in hostile # search still ran with that query
|
||||
assert hostile.split("<title>", 1)[1].split("</title>", 1)[0] != "PWNED" # escaped text, not run
|
||||
assert "data-pwned" not in hostile
|
||||
@@ -0,0 +1,260 @@
|
||||
"""Follow-up security fixes for the residual Python CodeQL lane (PR #6503).
|
||||
|
||||
A. upload-reference scanners stay linear and match the regexes they replaced.
|
||||
B. /api/hwfit/* (SSH host probing, host path probing) is admin-only.
|
||||
C. IMAP/SMTP connections honour the outbound address policy, and the account
|
||||
connection test never echoes a non-mail service's bytes.
|
||||
D. attachment extraction directories are per owner/account, not per folder/UID.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import imaplib
|
||||
import re
|
||||
import smtplib
|
||||
import socket
|
||||
import ssl
|
||||
import time
|
||||
from email.mime.application import MIMEApplication
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.text import MIMEText
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
# -- A. upload reference scanners ---------------------------------------------
|
||||
|
||||
_HEX = "a" * 32
|
||||
_PDF_REF = re.compile(
|
||||
r"<!--\s*pdf(?:_form)?_source\b[^>]*\bupload_id=[\"']([0-9a-fA-F]{32}(?:\.[A-Za-z0-9]+)?)[\"'][^>]*-->",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_ATTACHMENT_REF = re.compile(
|
||||
r"\[Attachment:[^\]\r\n]*\|\s*id=([0-9a-fA-F]{32}(?:\.[A-Za-z0-9]+)?)(?:\s*\||\s*\])",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("text", [
|
||||
f"<!-- pdf_source upload_id='{_HEX}' -->",
|
||||
f"<!--PDF_FORM_SOURCE x upload_id=\"{_HEX}.pdf\" y-->",
|
||||
f"<!-- pdf_source upload_id='{_HEX}' upload_id='{'b' * 32}' -->",
|
||||
f"<!-- pdf_source upload_id='{_HEX}' ->",
|
||||
f"<!-- pdf_sourceupload_id='{_HEX}' -->",
|
||||
f"<!--pdf_source <!--pdf_source upload_id='{_HEX}'-->",
|
||||
"<!-- pdf_source no id -->",
|
||||
])
|
||||
def test_pdf_source_ids_match_reference(text):
|
||||
from src.upload_handler import _pdf_source_upload_ids
|
||||
|
||||
assert _pdf_source_upload_ids(text) == _PDF_REF.findall(text)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("text", [
|
||||
f"[Attachment: report.pdf | id={_HEX} | 12 KB]",
|
||||
f"[attachment: a | b | id={_HEX}.txt]",
|
||||
f"[Attachment: a | id={_HEX} | id={'b' * 32}]",
|
||||
f"[Attachment: x |\nid={_HEX}]",
|
||||
f"[Attachment: [Attachment: n | id={_HEX}]",
|
||||
f"[Attachment: n | id={_HEX}x]",
|
||||
"[Attachment: no id here]",
|
||||
])
|
||||
def test_attachment_reference_ids_match_reference(text):
|
||||
from src.upload_handler import _attachment_reference_ids
|
||||
|
||||
assert _attachment_reference_ids(text) == _ATTACHMENT_REF.findall(text)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("flood", [
|
||||
"[attachment:" * 20_000, # CodeQL's reported shape (#615)
|
||||
"[Attachment: a |" * 15_000,
|
||||
"<!-- pdf_source upload_id=" * 9_000, # same sink, other pattern
|
||||
"<!--pdf_source " + f"upload_id='{_HEX}' " * 5_000 + "-->",
|
||||
])
|
||||
def test_internal_upload_id_extraction_is_linear(flood):
|
||||
from src.upload_handler import extract_internal_upload_ids
|
||||
|
||||
started = time.perf_counter()
|
||||
extract_internal_upload_ids(flood)
|
||||
assert time.perf_counter() - started < 2.0
|
||||
|
||||
|
||||
# -- B. hwfit authorization -----------------------------------------------------
|
||||
|
||||
def _hwfit_client(monkeypatch, *, auth_disabled=False):
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.testclient import TestClient
|
||||
import core.middleware as middleware
|
||||
import services.hwfit.hardware as hardware
|
||||
from routes.hwfit_routes import setup_hwfit_routes
|
||||
|
||||
probes = []
|
||||
monkeypatch.setattr(middleware, "auth_disabled", lambda: auth_disabled)
|
||||
monkeypatch.setattr(
|
||||
hardware, "detect_system",
|
||||
lambda **kwargs: probes.append(kwargs) or {"backend": "cpu_x86"},
|
||||
)
|
||||
app = FastAPI()
|
||||
app.state.auth_manager = SimpleNamespace(is_configured=True, is_admin=lambda user: user == "root")
|
||||
|
||||
@app.middleware("http")
|
||||
async def _user(request: Request, call_next):
|
||||
request.state.current_user = request.headers.get("x-test-user")
|
||||
return await call_next(request)
|
||||
|
||||
app.include_router(setup_hwfit_routes())
|
||||
return TestClient(app), probes
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", [
|
||||
"/api/hwfit/system?host=10.0.0.5&ssh_port=2222",
|
||||
"/api/hwfit/models?host=169.254.169.254",
|
||||
"/api/hwfit/profiles?model_path=/etc&host=",
|
||||
"/api/hwfit/image-models?host=internal-db",
|
||||
])
|
||||
def test_hwfit_rejects_non_admin_before_any_probe(monkeypatch, path):
|
||||
client, probes = _hwfit_client(monkeypatch)
|
||||
response = client.get(path, headers={"x-test-user": "alice"})
|
||||
assert response.status_code == 403
|
||||
assert probes == []
|
||||
|
||||
|
||||
def test_hwfit_allows_admin_and_single_user(monkeypatch):
|
||||
client, probes = _hwfit_client(monkeypatch)
|
||||
assert client.get("/api/hwfit/system", headers={"x-test-user": "root"}).status_code == 200
|
||||
client, probes = _hwfit_client(monkeypatch, auth_disabled=True)
|
||||
assert client.get("/api/hwfit/system", headers={"x-test-user": "anyone"}).status_code == 200
|
||||
assert probes
|
||||
|
||||
|
||||
# -- C. mail server address policy and test-connection errors -------------------
|
||||
|
||||
def test_imap_and_smtp_helpers_refuse_metadata_hosts_before_any_socket(monkeypatch):
|
||||
"""Real helper paths (no stdlib patching): the policy rejects a literal
|
||||
metadata address before a socket is even created. Policy matrix and DNS
|
||||
rebinding coverage live in test_codeql_python_residual_closure.py."""
|
||||
import routes.email.email_helpers as helpers
|
||||
|
||||
monkeypatch.delenv("EMAIL_BLOCK_PRIVATE_IPS", raising=False)
|
||||
with pytest.raises(helpers.MailServerAddressBlocked):
|
||||
helpers._open_imap_connection("169.254.169.254", 993, starttls=False, owner="alice")
|
||||
with pytest.raises(helpers.MailServerAddressBlocked):
|
||||
helpers._send_smtp_message(
|
||||
{"smtp_host": "169.254.169.254", "smtp_port": 587, "owner": "alice"}, "a@b", ["c@d"], "x",
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("error,expected,leak", [
|
||||
(imaplib.IMAP4.abort("unexpected response: b'SSH-2.0-OpenSSH_9.6 internal-db'"),
|
||||
"IMAP server did not respond like an IMAP server", "OpenSSH"),
|
||||
(smtplib.SMTPConnectError(554, b"-ERR redis 7.2 internal-cache"),
|
||||
"SMTP server did not respond like an SMTP server", "redis"),
|
||||
(smtplib.SMTPServerDisconnected("Connection unexpectedly closed: HTTP/1.1 400"),
|
||||
"SMTP server did not respond like an SMTP server", "HTTP/1.1"),
|
||||
(ConnectionRefusedError(111, "Connection refused 10.0.0.5:6379"), "IMAP connection refused", "6379"),
|
||||
(socket.timeout("timed out"), "IMAP connection timed out", "10.0"),
|
||||
(ssl.SSLError(1, "[SSL: WRONG_VERSION_NUMBER] banner: nginx"),
|
||||
"IMAP TLS handshake failed; check the port and security setting", "nginx"),
|
||||
])
|
||||
def test_connection_test_errors_never_echo_peer_bytes(error, expected, leak):
|
||||
from routes.email.email_helpers import _mail_connection_test_error
|
||||
|
||||
protocol = expected.split()[0]
|
||||
message = _mail_connection_test_error(protocol, "10.0.0.5", error)
|
||||
assert message == expected
|
||||
assert leak not in message
|
||||
|
||||
|
||||
def test_connection_test_keeps_mail_server_auth_responses():
|
||||
from routes.email.email_helpers import _mail_connection_test_error
|
||||
|
||||
assert _mail_connection_test_error(
|
||||
"IMAP", "imap.example.com", imaplib.IMAP4.error("[AUTHENTICATIONFAILED] Invalid credentials"),
|
||||
) == "[AUTHENTICATIONFAILED] Invalid credentials"
|
||||
microsoft = _mail_connection_test_error(
|
||||
"SMTP", "smtp.office365.com",
|
||||
smtplib.SMTPAuthenticationError(535, b"5.7.139 Authentication unsuccessful"),
|
||||
)
|
||||
assert "Microsoft no longer accepts" in microsoft
|
||||
|
||||
|
||||
def test_account_connection_test_route_applies_policy_to_both_protocols(monkeypatch):
|
||||
import routes.email_routes as email_routes
|
||||
|
||||
monkeypatch.delenv("EMAIL_BLOCK_PRIVATE_IPS", raising=False)
|
||||
router = email_routes.setup_email_routes()
|
||||
endpoint = next(
|
||||
r.endpoint for r in router.routes
|
||||
if r.path == "/api/email/accounts/test" and "POST" in getattr(r, "methods", set())
|
||||
)
|
||||
|
||||
class _Request:
|
||||
async def json(self):
|
||||
return {
|
||||
"imap_host": "169.254.169.254", "imap_port": 993, "imap_user": "u", "imap_password": "p",
|
||||
"smtp_host": "169.254.169.254", "smtp_port": 587,
|
||||
}
|
||||
|
||||
result = asyncio.run(endpoint(req=_Request(), owner="alice"))
|
||||
policy = "server address is not allowed by this server's network policy"
|
||||
assert result["ok"] is False
|
||||
assert result["imap"]["error"] == f"IMAP {policy}"
|
||||
assert result["smtp"]["error"] == f"SMTP {policy}"
|
||||
|
||||
|
||||
# -- D. per-owner/account attachment extraction -------------------------------
|
||||
|
||||
def _message_with_attachment(body: bytes) -> bytes:
|
||||
msg = MIMEMultipart()
|
||||
msg.attach(MIMEText("see attached", "plain"))
|
||||
part = MIMEApplication(body, Name="invoice.pdf")
|
||||
part["Content-Disposition"] = 'attachment; filename="invoice.pdf"'
|
||||
msg.attach(part)
|
||||
return msg.as_bytes()
|
||||
|
||||
|
||||
def test_attachment_extract_dir_separates_owners_and_accounts(tmp_path, monkeypatch):
|
||||
import routes.email.email_helpers as helpers
|
||||
|
||||
monkeypatch.setattr(helpers, "ATTACHMENTS_DIR", tmp_path)
|
||||
alice = helpers.attachment_extract_dir("INBOX", "42", owner="alice", account_id="a1")
|
||||
assert alice != helpers.attachment_extract_dir("INBOX", "42", owner="bob", account_id="a1")
|
||||
assert alice != helpers.attachment_extract_dir("INBOX", "42", owner="alice", account_id="a2")
|
||||
assert alice != helpers.attachment_extract_dir("INBOX/42", "", owner="alice", account_id="a1")
|
||||
assert alice.parent == tmp_path.resolve()
|
||||
|
||||
|
||||
def test_two_users_same_folder_uid_cannot_overwrite_each_others_attachment(tmp_path, monkeypatch):
|
||||
"""Two principals, same INBOX UID 42, same attachment filename.
|
||||
|
||||
Before the fix both extractions landed in ATTACHMENTS_DIR/INBOX_42/, so
|
||||
bob's request rewrote the file whose path alice's agent was about to read.
|
||||
"""
|
||||
import routes.email.email_helpers as helpers
|
||||
import routes.email_routes as email_routes
|
||||
|
||||
monkeypatch.setattr(helpers, "ATTACHMENTS_DIR", tmp_path)
|
||||
mailboxes = {
|
||||
"alice": _message_with_attachment(b"%PDF alice private invoice"),
|
||||
"bob": _message_with_attachment(b"%PDF bob controlled content"),
|
||||
}
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _fake_imap(account_id=None, owner=""):
|
||||
yield SimpleNamespace(owner=owner, select=lambda *a, **k: ("OK", [b"1"]))
|
||||
|
||||
monkeypatch.setattr(email_routes, "_imap", _fake_imap)
|
||||
monkeypatch.setattr(
|
||||
email_routes, "_imap_uid_fetch",
|
||||
lambda conn, uid, query: ("OK", [(b"42 (RFC822)", mailboxes[conn.owner])]),
|
||||
)
|
||||
router = email_routes.setup_email_routes()
|
||||
endpoint = next(r.endpoint for r in router.routes if r.path == "/api/email/attachment-path/{uid}/{index}")
|
||||
|
||||
alice = asyncio.run(endpoint(uid="42", index=0, folder="INBOX", account_id=None, owner="alice"))
|
||||
bob = asyncio.run(endpoint(uid="42", index=0, folder="INBOX", account_id=None, owner="bob"))
|
||||
|
||||
assert alice["filename"] == bob["filename"] == "invoice.pdf"
|
||||
assert alice["path"] != bob["path"]
|
||||
with open(alice["path"], "rb") as handle:
|
||||
assert handle.read() == b"%PDF alice private invoice"
|
||||
@@ -0,0 +1,164 @@
|
||||
"""Adversarial tests for the residual Python CodeQL fixes (PR #6503 lane).
|
||||
|
||||
Covers py/reflective-xss (search page), py/path-injection (context_info host
|
||||
path oracle) and py/stack-trace-exposure (readiness, stream errors, task
|
||||
drafting).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# -- py/reflective-xss: GET /search/web?q= ----------------------------------
|
||||
|
||||
_XSS_QUERIES = [
|
||||
'</script><script src="https://cdn.jsdelivr.net/gh/evil/x@1/p.js"></script>',
|
||||
"</SCRIPT><img src=x onerror=alert(1)>",
|
||||
"<!--<script>",
|
||||
"a & b < c > d
\"quoted\" 'single'",
|
||||
]
|
||||
|
||||
|
||||
def _search_page(q):
|
||||
from routes.search.search_routes import setup_search_routes
|
||||
|
||||
router = setup_search_routes(SimpleNamespace())
|
||||
endpoint = next(r.endpoint for r in router.routes if r.path == "/search/web")
|
||||
request = SimpleNamespace(state=SimpleNamespace(csp_nonce="n0nce"))
|
||||
return asyncio.run(endpoint(request=request, q=q)).body.decode("utf-8")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("q", _XSS_QUERIES)
|
||||
def test_search_page_query_cannot_break_out_of_inline_script(q):
|
||||
page = _search_page(q)
|
||||
script = page.split('<script nonce="n0nce">', 1)[1]
|
||||
# The page's own closing tag is the only one; nothing from q survives raw.
|
||||
assert script.lower().count("</script") == 1
|
||||
assert "<!--" not in script
|
||||
literal = re.search(r"const initialQuery = (.*?);\n", script).group(1)
|
||||
assert "<" not in literal and ">" not in literal and "&" not in literal
|
||||
# The JS string value is unchanged for the search itself.
|
||||
assert json.loads(literal) == q.strip()
|
||||
|
||||
|
||||
# -- py/path-injection: /session/{id}/context_info?cwd= ----------------------
|
||||
|
||||
def _context_info(monkeypatch, *, admin, cwd):
|
||||
import routes.session_routes as routes
|
||||
from fastapi import APIRouter
|
||||
|
||||
monkeypatch.setattr(routes, "_verify_session_owner", lambda *args: None)
|
||||
monkeypatch.setattr("src.tool_security.blocked_tools_for_owner", lambda owner: set())
|
||||
monkeypatch.setattr("src.tool_security.owner_is_admin_or_single_user", lambda owner: admin)
|
||||
monkeypatch.setattr(routes, "router", APIRouter(prefix="/api"))
|
||||
session = SimpleNamespace(endpoint_url="", model="", cwd="")
|
||||
router = routes.setup_session_routes(SimpleNamespace(get_session=lambda sid: session), {})
|
||||
endpoint = next(
|
||||
r.endpoint for r in router.routes if r.path == "/api/session/{session_id}/context_info"
|
||||
)
|
||||
request = SimpleNamespace(state=SimpleNamespace(
|
||||
api_token=False, api_token_owner=None, api_token_scopes=[], current_user="someone",
|
||||
))
|
||||
return asyncio.run(endpoint(request, "session-1", cwd=cwd))
|
||||
|
||||
|
||||
def test_context_info_hides_host_paths_from_non_admin(monkeypatch, tmp_path):
|
||||
(tmp_path / "AGENTS.md").write_text("x", encoding="utf-8")
|
||||
result = _context_info(monkeypatch, admin=False, cwd=str(tmp_path))
|
||||
assert result["agents_md"] == []
|
||||
assert result["workspace"]["exists_in_backend"] is False
|
||||
|
||||
|
||||
def test_context_info_still_reports_host_paths_for_admin(monkeypatch, tmp_path):
|
||||
(tmp_path / "AGENTS.md").write_text("x", encoding="utf-8")
|
||||
result = _context_info(monkeypatch, admin=True, cwd=str(tmp_path))
|
||||
assert {"path": str(tmp_path / "AGENTS.md"), "source": "workspace"} in result["agents_md"]
|
||||
assert result["workspace"]["exists_in_backend"] is True
|
||||
|
||||
|
||||
# -- py/stack-trace-exposure ---------------------------------------------------
|
||||
|
||||
_SECRET = "postgresql://odysseus:hunter2@10.0.0.5/prod /srv/odysseus/data"
|
||||
|
||||
|
||||
def test_readiness_does_not_return_raw_database_error(monkeypatch):
|
||||
import core.database as database
|
||||
from src.readiness import check_readiness
|
||||
|
||||
class _Engine:
|
||||
def connect(self):
|
||||
raise RuntimeError(_SECRET)
|
||||
|
||||
monkeypatch.setattr(database, "engine", _Engine())
|
||||
result = check_readiness()
|
||||
db_check = result["checks"]["database"]
|
||||
assert db_check == {"ok": False, "error_type": "RuntimeError"}
|
||||
assert "hunter2" not in json.dumps(result)
|
||||
assert result["ready"] is False
|
||||
|
||||
|
||||
def test_readiness_does_not_return_raw_data_dir_error(monkeypatch, tmp_path):
|
||||
import core.constants as constants
|
||||
from src.readiness import check_readiness
|
||||
|
||||
blocker = tmp_path / "not-a-dir"
|
||||
blocker.write_text("file, not a directory", encoding="utf-8")
|
||||
monkeypatch.setattr(constants, "DATA_DIR", str(blocker / "data"))
|
||||
result = check_readiness()
|
||||
data_check = result["checks"]["data_dir"]
|
||||
assert data_check["ok"] is False
|
||||
assert set(data_check) == {"ok", "error_type"}
|
||||
assert str(tmp_path) not in json.dumps(result)
|
||||
|
||||
|
||||
def test_stream_failure_message_is_generic():
|
||||
from src.llm_core import _stream_failure_message
|
||||
|
||||
assert _stream_failure_message(RuntimeError(_SECRET)) == "Model stream failed (RuntimeError)"
|
||||
|
||||
|
||||
def test_catch_all_stream_error_events_do_not_embed_exception_text():
|
||||
from pathlib import Path
|
||||
|
||||
source = (Path(__file__).resolve().parents[1] / "src/llm_core.py").read_text(encoding="utf-8")
|
||||
assert '{"error": str(e), "status": 502' not in source
|
||||
|
||||
|
||||
def _parse_task(monkeypatch, exc):
|
||||
from fastapi import HTTPException # noqa: F401 (raised by callers below)
|
||||
from unittest.mock import MagicMock
|
||||
import routes.task.task_routes as task_routes
|
||||
import src.endpoint_resolver as resolver
|
||||
import src.llm_core as llm_core
|
||||
|
||||
async def _boom(*args, **kwargs):
|
||||
raise exc
|
||||
|
||||
monkeypatch.setattr(resolver, "resolve_endpoint", lambda *a, **k: ("http://llm", "m", {}))
|
||||
monkeypatch.setattr(llm_core, "llm_call_async", _boom)
|
||||
router = task_routes.setup_task_routes(MagicMock())
|
||||
endpoint = next(r.endpoint for r in router.routes if r.path == "/api/tasks/parse")
|
||||
|
||||
class _Request:
|
||||
state = SimpleNamespace(current_user="alice")
|
||||
|
||||
async def json(self):
|
||||
return {"description": "every day at 9 summarize the news"}
|
||||
|
||||
return asyncio.run(endpoint(_Request()))
|
||||
|
||||
|
||||
def test_parse_task_hides_unexpected_exception_text(monkeypatch):
|
||||
result = _parse_task(monkeypatch, RuntimeError(_SECRET))
|
||||
assert result == {"success": False, "message": "Could not draft a task (RuntimeError)"}
|
||||
|
||||
|
||||
def test_parse_task_keeps_curated_upstream_http_error(monkeypatch):
|
||||
from fastapi import HTTPException
|
||||
|
||||
result = _parse_task(monkeypatch, HTTPException(401, "Upstream rejected the API key"))
|
||||
assert result == {"success": False, "message": "Upstream rejected the API key"}
|
||||
@@ -26,7 +26,7 @@ class _FakeSock:
|
||||
class _FakeIMAP:
|
||||
calls = []
|
||||
|
||||
def __init__(self, host, port, timeout=None):
|
||||
def __init__(self, host, port, timeout=None, **_policy):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.timeout = timeout
|
||||
@@ -60,8 +60,8 @@ def test_open_imap_connection_uses_shared_timeout_for_implicit_ssl(monkeypatch):
|
||||
import routes.email_helpers as helpers
|
||||
|
||||
_FakeIMAP.calls = []
|
||||
monkeypatch.setattr(helpers.imaplib, "IMAP4", _FakeIMAP)
|
||||
monkeypatch.setattr(helpers.imaplib, "IMAP4_SSL", _FakeIMAPSSL)
|
||||
monkeypatch.setattr(helpers, "_PolicyIMAP4", _FakeIMAP)
|
||||
monkeypatch.setattr(helpers, "_PolicyIMAP4_SSL", _FakeIMAPSSL)
|
||||
|
||||
conn = _open_imap_connection("imap.one.com", 993, starttls=False)
|
||||
|
||||
@@ -75,8 +75,8 @@ def test_open_imap_connection_supports_starttls(monkeypatch):
|
||||
import routes.email_helpers as helpers
|
||||
|
||||
_FakeIMAP.calls = []
|
||||
monkeypatch.setattr(helpers.imaplib, "IMAP4", _FakeIMAP)
|
||||
monkeypatch.setattr(helpers.imaplib, "IMAP4_SSL", _FakeIMAPSSL)
|
||||
monkeypatch.setattr(helpers, "_PolicyIMAP4", _FakeIMAP)
|
||||
monkeypatch.setattr(helpers, "_PolicyIMAP4_SSL", _FakeIMAPSSL)
|
||||
|
||||
_open_imap_connection("imap.local", 143, starttls=True)
|
||||
|
||||
@@ -99,7 +99,7 @@ async def test_account_config_uses_shared_imap_timeout(monkeypatch):
|
||||
def logout(self):
|
||||
captured["logout"] = True
|
||||
|
||||
def fake_open(host, port, *, starttls, timeout):
|
||||
def fake_open(host, port, *, starttls, timeout, **_policy):
|
||||
captured["open"] = (host, port, starttls, timeout)
|
||||
return _Conn()
|
||||
|
||||
|
||||
@@ -64,7 +64,7 @@ def test_imap_connect_with_host_still_connects(monkeypatch):
|
||||
def login(self, user, password):
|
||||
opened["login"] = (user, password)
|
||||
|
||||
def _fake_open(host, port, *, starttls, timeout):
|
||||
def _fake_open(host, port, *, starttls, timeout, **_policy):
|
||||
opened["host"] = host
|
||||
return _FakeConn()
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@ from routes.email_helpers import _send_smtp_message
|
||||
class _FakeSMTP:
|
||||
calls = []
|
||||
|
||||
def __init__(self, host, port, timeout=None):
|
||||
def __init__(self, host, port, timeout=None, **_policy):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.timeout = timeout
|
||||
@@ -54,8 +54,8 @@ def test_send_smtp_message_supports_plain_smtp(monkeypatch):
|
||||
import routes.email_helpers as helpers
|
||||
|
||||
_FakeSMTP.calls = []
|
||||
monkeypatch.setattr(helpers.smtplib, "SMTP", _FakeSMTP)
|
||||
monkeypatch.setattr(helpers.smtplib, "SMTP_SSL", _FakeSMTPSSL)
|
||||
monkeypatch.setattr(helpers, "_PolicySMTP", _FakeSMTP)
|
||||
monkeypatch.setattr(helpers, "_PolicySMTP_SSL", _FakeSMTPSSL)
|
||||
|
||||
_send_smtp_message(_cfg("none"), "from@example.com", ["to@example.com"], "hello")
|
||||
|
||||
@@ -68,8 +68,8 @@ def test_send_smtp_message_supports_explicit_starttls(monkeypatch):
|
||||
import routes.email_helpers as helpers
|
||||
|
||||
_FakeSMTP.calls = []
|
||||
monkeypatch.setattr(helpers.smtplib, "SMTP", _FakeSMTP)
|
||||
monkeypatch.setattr(helpers.smtplib, "SMTP_SSL", _FakeSMTPSSL)
|
||||
monkeypatch.setattr(helpers, "_PolicySMTP", _FakeSMTP)
|
||||
monkeypatch.setattr(helpers, "_PolicySMTP_SSL", _FakeSMTPSSL)
|
||||
|
||||
_send_smtp_message(_cfg("starttls", port=2525), "from@example.com", ["to@example.com"], "hello")
|
||||
|
||||
@@ -82,8 +82,8 @@ def test_send_smtp_message_defaults_587_to_starttls(monkeypatch):
|
||||
import routes.email_helpers as helpers
|
||||
|
||||
_FakeSMTP.calls = []
|
||||
monkeypatch.setattr(helpers.smtplib, "SMTP", _FakeSMTP)
|
||||
monkeypatch.setattr(helpers.smtplib, "SMTP_SSL", _FakeSMTPSSL)
|
||||
monkeypatch.setattr(helpers, "_PolicySMTP", _FakeSMTP)
|
||||
monkeypatch.setattr(helpers, "_PolicySMTP_SSL", _FakeSMTPSSL)
|
||||
|
||||
cfg = _cfg("", port=587)
|
||||
_send_smtp_message(cfg, "from@example.com", ["to@example.com"], "hello")
|
||||
@@ -96,8 +96,8 @@ def test_send_smtp_message_uses_ssl_when_configured(monkeypatch):
|
||||
import routes.email_helpers as helpers
|
||||
|
||||
_FakeSMTP.calls = []
|
||||
monkeypatch.setattr(helpers.smtplib, "SMTP", _FakeSMTP)
|
||||
monkeypatch.setattr(helpers.smtplib, "SMTP_SSL", _FakeSMTPSSL)
|
||||
monkeypatch.setattr(helpers, "_PolicySMTP", _FakeSMTP)
|
||||
monkeypatch.setattr(helpers, "_PolicySMTP_SSL", _FakeSMTPSSL)
|
||||
|
||||
_send_smtp_message(_cfg("ssl", port=465), "from@example.com", ["to@example.com"], "hello")
|
||||
|
||||
|
||||
@@ -89,8 +89,8 @@ async def test_test_connection_oauth_account_uses_xoauth2_for_imap_and_smtp():
|
||||
|
||||
with mock.patch("core.database.SessionLocal", Factory), \
|
||||
mock.patch("routes.email_routes._open_imap_connection", return_value=mock_imap_conn), \
|
||||
mock.patch("routes.email_routes.smtplib.SMTP", return_value=mock_smtp_conn), \
|
||||
mock.patch("routes.email_routes.smtplib.SMTP_SSL", return_value=mock_smtp_conn), \
|
||||
mock.patch("routes.email_routes._PolicySMTP", return_value=mock_smtp_conn), \
|
||||
mock.patch("routes.email_routes._PolicySMTP_SSL", return_value=mock_smtp_conn), \
|
||||
mock.patch("routes.email_routes._get_valid_google_token", return_value="ya29.live") as token_getter:
|
||||
result = await test_conn(req=_FakeReq(), owner="alice")
|
||||
|
||||
@@ -141,8 +141,8 @@ async def test_test_connection_password_account_still_uses_login():
|
||||
|
||||
with mock.patch("core.database.SessionLocal", Factory), \
|
||||
mock.patch("routes.email_routes._open_imap_connection", return_value=mock_imap_conn), \
|
||||
mock.patch("routes.email_routes.smtplib.SMTP", return_value=mock_smtp_conn), \
|
||||
mock.patch("routes.email_routes.smtplib.SMTP_SSL", return_value=mock_smtp_conn):
|
||||
mock.patch("routes.email_routes._PolicySMTP", return_value=mock_smtp_conn), \
|
||||
mock.patch("routes.email_routes._PolicySMTP_SSL", return_value=mock_smtp_conn):
|
||||
result = await test_conn(req=_FakeReq(), owner="alice")
|
||||
|
||||
assert result["ok"] is True
|
||||
@@ -187,8 +187,8 @@ async def test_test_connection_rejects_non_google_hosts_before_oauth_auth():
|
||||
|
||||
with mock.patch("core.database.SessionLocal", Factory), \
|
||||
mock.patch("routes.email_routes._open_imap_connection") as open_imap, \
|
||||
mock.patch("routes.email_routes.smtplib.SMTP") as open_smtp, \
|
||||
mock.patch("routes.email_routes.smtplib.SMTP_SSL") as open_smtp_ssl, \
|
||||
mock.patch("routes.email_routes._PolicySMTP") as open_smtp, \
|
||||
mock.patch("routes.email_routes._PolicySMTP_SSL") as open_smtp_ssl, \
|
||||
mock.patch("routes.email_routes._get_valid_google_token") as token_getter:
|
||||
result = await test_conn(req=_FakeReq(), owner="alice")
|
||||
|
||||
@@ -238,8 +238,8 @@ async def test_test_connection_rejects_insecure_oauth_transports_before_auth():
|
||||
|
||||
with mock.patch("core.database.SessionLocal", Factory), \
|
||||
mock.patch("routes.email_routes._open_imap_connection") as open_imap, \
|
||||
mock.patch("routes.email_routes.smtplib.SMTP") as open_smtp, \
|
||||
mock.patch("routes.email_routes.smtplib.SMTP_SSL") as open_smtp_ssl, \
|
||||
mock.patch("routes.email_routes._PolicySMTP") as open_smtp, \
|
||||
mock.patch("routes.email_routes._PolicySMTP_SSL") as open_smtp_ssl, \
|
||||
mock.patch("routes.email_routes._get_valid_google_token") as token_getter:
|
||||
result = await test_conn(req=_FakeReq(), owner="alice")
|
||||
|
||||
@@ -336,10 +336,10 @@ async def test_test_connection_verifies_imap_tls_before_loading_oauth_token(
|
||||
"routes.email_routes.ssl.create_default_context",
|
||||
return_value=context,
|
||||
), mock.patch(
|
||||
"routes.email_helpers.imaplib.IMAP4",
|
||||
"routes.email_helpers._PolicyIMAP4",
|
||||
return_value=starttls_conn,
|
||||
) as imap_cls, mock.patch(
|
||||
"routes.email_helpers.imaplib.IMAP4_SSL",
|
||||
"routes.email_helpers._PolicyIMAP4_SSL",
|
||||
side_effect=ssl.SSLCertVerificationError("untrusted certificate"),
|
||||
) as imap_ssl_cls, mock.patch(
|
||||
"routes.email_routes._get_valid_google_token"
|
||||
@@ -412,10 +412,10 @@ async def test_test_connection_verifies_smtp_tls_before_loading_oauth_token(
|
||||
"routes.email_routes.ssl.create_default_context",
|
||||
return_value=context,
|
||||
), mock.patch(
|
||||
"routes.email_routes.smtplib.SMTP",
|
||||
"routes.email_routes._PolicySMTP",
|
||||
return_value=starttls_smtp,
|
||||
) as smtp_cls, mock.patch(
|
||||
"routes.email_routes.smtplib.SMTP_SSL",
|
||||
"routes.email_routes._PolicySMTP_SSL",
|
||||
side_effect=ssl.SSLCertVerificationError("untrusted certificate"),
|
||||
) as smtp_ssl_cls, mock.patch(
|
||||
"routes.email_routes._get_valid_google_token"
|
||||
|
||||
@@ -281,7 +281,7 @@ def test_open_imap_connection_shuts_down_on_starttls_failure(monkeypatch):
|
||||
))
|
||||
conn.starttls = MagicMock(side_effect=RuntimeError("STARTTLS rejected"))
|
||||
|
||||
monkeypatch.setattr(helpers.imaplib, "IMAP4", lambda *a, **kw: conn)
|
||||
monkeypatch.setattr(helpers, "_PolicyIMAP4", lambda *a, **kw: conn)
|
||||
|
||||
raised = False
|
||||
try:
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
"""manage_notes reminder-time sniffing must stay linear (CodeQL py/polynomial-redos).
|
||||
|
||||
`\\d{1,2}(?::\\d{2})?\\s*(?:am|pm)?\\s+today` let `\\s*` and `\\s+` split one
|
||||
whitespace run every possible way, so a digit followed by a long whitespace
|
||||
run in model-supplied note text was O(n^2). The rewrite groups the optional
|
||||
meridiem with its own whitespace: same language, one way to match.
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.helpers.import_state import clear_fake_database_modules
|
||||
from tests.helpers.sqlite_db import make_temp_sqlite
|
||||
|
||||
clear_fake_database_modules()
|
||||
|
||||
import core.database as cdb
|
||||
|
||||
_TS, _ENGINE, _TMPDB = make_temp_sqlite(cdb.Base.metadata)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _bind_temp_db(monkeypatch):
|
||||
monkeypatch.setitem(sys.modules, "core.database", cdb)
|
||||
parent = sys.modules.get("core")
|
||||
if parent is not None:
|
||||
monkeypatch.setattr(parent, "database", cdb, raising=False)
|
||||
monkeypatch.setattr(cdb, "SessionLocal", _TS)
|
||||
yield
|
||||
|
||||
|
||||
async def _add(content):
|
||||
from src.tool_implementations import do_manage_notes
|
||||
|
||||
return await do_manage_notes(
|
||||
json.dumps({"action": "add", "title": "remind me", "content": content}),
|
||||
owner="tester-" + uuid.uuid4().hex[:6],
|
||||
)
|
||||
|
||||
|
||||
async def test_reminder_time_sniffing_whitespace_flood_is_linear():
|
||||
started = time.perf_counter()
|
||||
res = await _add("remind 1" + "\t" * 40_000 + "x")
|
||||
assert time.perf_counter() - started < 2.0
|
||||
assert res.get("exit_code", 0) == 0, res
|
||||
|
||||
|
||||
async def test_reminder_time_sniffing_still_finds_time_first_phrases():
|
||||
res = await _add("remind me to call mom, 9 pm tomorrow")
|
||||
assert res.get("exit_code", 0) == 0, res
|
||||
db = _TS()
|
||||
try:
|
||||
note = db.query(cdb.Note).filter(cdb.Note.id == res["note_id"]).first()
|
||||
assert note is not None and "T21:00:00" in (note.due_date or ""), note and note.due_date
|
||||
finally:
|
||||
db.close()
|
||||
@@ -0,0 +1,183 @@
|
||||
"""Regression tests for the residual py/polynomial-redos fixes (PR #6503 lane).
|
||||
|
||||
Each rewritten matcher is pinned two ways:
|
||||
* equivalence with the regex it replaced on ordinary inputs, and
|
||||
* the CodeQL-reported adversarial shape completing within a loose budget
|
||||
(the replaced regexes took seconds to tens of seconds on these sizes).
|
||||
"""
|
||||
|
||||
import re
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.text_helpers import strip_closed_think_blocks
|
||||
|
||||
_BUDGET_S = 2.0
|
||||
_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def _fast(fn, *args):
|
||||
started = time.perf_counter()
|
||||
result = fn(*args)
|
||||
elapsed = time.perf_counter() - started
|
||||
assert elapsed < _BUDGET_S, f"{getattr(fn, '__name__', fn)} took {elapsed:.2f}s"
|
||||
return result
|
||||
|
||||
|
||||
# -- closed <think> blocks (skills_routes x4, memory_extractor.audit_memories) --
|
||||
|
||||
_THINK_REF = re.compile(r"<think(?:ing)?>[\s\S]*?</think(?:ing)?>", re.I)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("text", [
|
||||
"",
|
||||
"plain",
|
||||
"<think>x</think>{\"ok\": true}",
|
||||
"a<thinking>b</thinking>c<think>d</think>e",
|
||||
"<think>a<think>nested</think>rest</think>",
|
||||
"orphan</think> then <think>unclosed",
|
||||
"<THINK>case</Thinking>kept",
|
||||
"<think>\nmulti\nline\n</think>\n{}",
|
||||
])
|
||||
def test_strip_closed_think_blocks_matches_reference(text):
|
||||
assert strip_closed_think_blocks(text) == _THINK_REF.sub("", text)
|
||||
|
||||
|
||||
def test_strip_closed_think_blocks_unclosed_opener_flood_is_linear():
|
||||
flood = "<think>" * 40_000
|
||||
assert _fast(strip_closed_think_blocks, flood) == flood
|
||||
|
||||
|
||||
def test_lazy_think_regex_is_gone_from_llm_review_parsers():
|
||||
for rel in ("routes/skills_routes.py", "services/memory/memory_extractor.py"):
|
||||
source = (_ROOT / rel).read_text(encoding="utf-8")
|
||||
assert r"<think(?:ing)?>[\s\S]*?</think(?:ing)?>" not in source, rel
|
||||
|
||||
|
||||
# -- email compose markdown links (email_routes._md_to_email_html) -----------
|
||||
|
||||
_LINK_REF = re.compile(r"\[([^\]]+)\]\((https?://[^)\s]+)\)")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("text", [
|
||||
"see [docs](https://example.com/a) and [b](http://x.y)",
|
||||
"[a[b](http://x)",
|
||||
"[]()[x](http://y)",
|
||||
"[a](http://x [b c](http://y)",
|
||||
"[a](http://)[b](https://z)",
|
||||
"[a](ftp://x) [b](https://ok)",
|
||||
"[unterminated(http://x)",
|
||||
"[t](http://u) tail ] [",
|
||||
])
|
||||
def test_md_links_to_html_matches_reference(text):
|
||||
from routes.email.email_routes import _md_links_to_html
|
||||
|
||||
assert _md_links_to_html(text) == _LINK_REF.sub(r'<a href="\2">\1</a>', text)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("flood", [
|
||||
"[" + "[\\" * 40_000, # CodeQL's reported shape
|
||||
"[a](http://x" * 8_000, # URL that never closes, repeated
|
||||
])
|
||||
def test_md_links_to_html_floods_are_linear(flood):
|
||||
from routes.email.email_routes import _md_links_to_html
|
||||
|
||||
assert _fast(_md_links_to_html, flood) == flood
|
||||
|
||||
|
||||
def test_md_to_email_html_still_renders_and_escapes_links():
|
||||
from routes.email.email_routes import _md_to_email_html
|
||||
|
||||
html = _md_to_email_html("**hi** [site](https://example.com/x)\n<script>")
|
||||
assert '<a href="https://example.com/x">site</a>' in html
|
||||
assert "<strong>hi</strong>" in html
|
||||
assert "<script>" not in html
|
||||
|
||||
|
||||
# -- PDF URL detection (chat_routes._prefers_structured_document_tools) ------
|
||||
|
||||
_PDF_REF = re.compile(r"https?://[^\s]+(?:\.pdf\b|/pdf/)", re.I)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("text", [
|
||||
"read https://arxiv.org/pdf/2401.0001 please",
|
||||
"https://x.org/paper.PDF",
|
||||
"https://x.org/paper.pdfx",
|
||||
"xhttps://a/b.pdf",
|
||||
"http:///pdf/",
|
||||
"http://a
.pdf",
|
||||
"no links here.pdf",
|
||||
"https://a/b http://c.pdf",
|
||||
])
|
||||
def test_mentions_pdf_url_matches_reference(text):
|
||||
from routes.chat_routes import _mentions_pdf_url
|
||||
|
||||
assert _mentions_pdf_url(text) is bool(_PDF_REF.search(text))
|
||||
|
||||
|
||||
def test_mentions_pdf_url_scheme_flood_is_linear():
|
||||
from routes.chat_routes import _prefers_structured_document_tools
|
||||
|
||||
assert _fast(_prefers_structured_document_tools, "http://" * 8_000) is False
|
||||
|
||||
|
||||
# -- chat_helpers persistence cleanup ----------------------------------------
|
||||
|
||||
def test_clean_repeated_assistant_content_trailing_think_closer():
|
||||
from routes.chat_helpers import clean_repeated_assistant_content
|
||||
|
||||
prose = "word " * 30 + "done" # past the 120-char edge-closer window
|
||||
assert clean_repeated_assistant_content(prose + " </think> ") == prose
|
||||
whitespace_run = "a" + "\t" * 80_000 + "x"
|
||||
assert _fast(clean_repeated_assistant_content, whitespace_run) == whitespace_run
|
||||
|
||||
|
||||
@pytest.mark.parametrize("text,expected", [
|
||||
("The user wants a hi.\n\n<think>Hey there!</think>",
|
||||
"<think>The user wants a hi.</think>\nHey there!"),
|
||||
("I should greet.\n<thinking> Hello ", "<think>I should greet.</think>\nHello"),
|
||||
("Plain answer <think>x</think>", "Plain answer <think>x</think>"),
|
||||
])
|
||||
def test_normalize_thinking_garbled_tags_unchanged(text, expected):
|
||||
from routes.chat_helpers import _normalize_thinking
|
||||
|
||||
assert _normalize_thinking(text) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize("flood", [
|
||||
"a" + "\n" * 80_000 + "x",
|
||||
"The user x\n<think>" + "\t" * 80_000 + "x",
|
||||
])
|
||||
def test_normalize_thinking_floods_are_linear(flood):
|
||||
from routes.chat_helpers import _normalize_thinking
|
||||
|
||||
_fast(_normalize_thinking, flood)
|
||||
|
||||
|
||||
# -- calendar natural-language datetimes (GET /events, POST /events) ---------
|
||||
|
||||
def test_calendar_time_first_phrases_still_parse():
|
||||
from routes.calendar_routes import _parse_dt, parse_due_for_user
|
||||
|
||||
parsed = _parse_dt("3pm tomorrow")
|
||||
assert (parsed.hour, parsed.minute) == (15, 0)
|
||||
assert "T09:00:00" in parse_due_for_user("9 a.m today")
|
||||
assert "T23:00:00" in parse_due_for_user("11pm tonight")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload", [
|
||||
"a" + "\t" * 80_000 + "x", # CodeQL's reported shape (time-first regex)
|
||||
"a" + "\t" * 80_000 + "m", # am/pm normaliser on the same input
|
||||
])
|
||||
def test_calendar_parsers_whitespace_flood_is_linear(payload):
|
||||
from routes.calendar_routes import _parse_dt, parse_due_for_user
|
||||
|
||||
for fn in (_parse_dt, parse_due_for_user):
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
fn(payload)
|
||||
except Exception:
|
||||
pass # unparseable is fine; only the time matters here
|
||||
assert time.perf_counter() - started < _BUDGET_S, fn.__name__
|
||||
@@ -984,7 +984,7 @@ def test_attachment_extract_dir_stays_contained(folder, uid):
|
||||
"""User-controlled folder/uid must never escape ATTACHMENTS_DIR — pins the
|
||||
fix for the attachment-extraction path traversal."""
|
||||
aed, base = _import_attachment_extract_dir()
|
||||
target = aed(folder, uid)
|
||||
target = aed(folder, uid, owner="../../owner", account_id="../acct")
|
||||
base_r = base.resolve()
|
||||
assert target == base_r or base_r in target.parents
|
||||
# exactly one extra path segment, and no `..` component survived
|
||||
@@ -992,9 +992,11 @@ def test_attachment_extract_dir_stays_contained(folder, uid):
|
||||
assert ".." not in rel.parts
|
||||
|
||||
|
||||
def test_attachment_extract_dir_normal_inputs_unchanged():
|
||||
def test_attachment_extract_dir_is_stable_single_segment_per_scope():
|
||||
aed, base = _import_attachment_extract_dir()
|
||||
assert aed("INBOX", "123") == base.resolve() / "INBOX_123"
|
||||
target = aed("INBOX", "123", owner="alice", account_id="acct-1")
|
||||
assert target == aed("INBOX", "123", owner="alice", account_id="acct-1")
|
||||
assert target.parent == base.resolve()
|
||||
|
||||
|
||||
def test_diagnostics_routes_are_admin_gated():
|
||||
|
||||
Reference in New Issue
Block a user