Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 19 additions & 2 deletions src/audit/crawler/js_fetcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

from __future__ import annotations

import asyncio
import contextlib
import io
from types import TracebackType
Expand Down Expand Up @@ -125,6 +126,7 @@ def __init__(
visual_probe: VisualProbe | None = None,
capture_screenshots: bool = False,
shared_context: BrowserContext | None = None,
shared_pages: tuple[Page, ...] = (),
private_context: bool = False,
max_rendered_html_chars: int | None = None,
headless: bool = True,
Expand Down Expand Up @@ -160,6 +162,14 @@ def __init__(
# context per page. Public scans keep the original isolated-context
# behavior.
self._shared_context = shared_context
if shared_pages and shared_context is None:
raise ValueError("shared_pages requires shared_context")
self._shared_pages = shared_pages
self._shared_page_pool: asyncio.Queue[Page] | None = None
if shared_pages:
self._shared_page_pool = asyncio.Queue(maxsize=len(shared_pages))
for shared_page in shared_pages:
self._shared_page_pool.put_nowait(shared_page)
# A protected companion must never put target paths, redirect URLs, or
# browser exception text (which can contain query/session material)
# into the application log. Public crawling keeps the useful existing
Expand Down Expand Up @@ -217,8 +227,13 @@ async def fetch(self, url: str) -> FetchResult:
)
owns_context = True
page: Page | None = None
pooled_page = False
try:
page = await ctx.new_page()
if self._shared_page_pool is not None:
page = await self._shared_page_pool.get()
pooled_page = True
else:
page = await ctx.new_page()
try:
resp = await page.goto(url, timeout=self._nav_timeout_ms, wait_until="load")
except Exception as exc:
Expand Down Expand Up @@ -357,7 +372,9 @@ async def fetch(self, url: str) -> FetchResult:
# wipes it only when the scan reaches a terminal state. Close the
# per-page tab, never the session itself. Public scans retain the
# old "new context per fetch" isolation guarantee.
if owns_context:
if pooled_page and page is not None and self._shared_page_pool is not None:
self._shared_page_pool.put_nowait(page)
elif owns_context:
await ctx.close()
elif page is not None:
with contextlib.suppress(Exception):
Expand Down
8 changes: 6 additions & 2 deletions src/audit/crawler/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -495,6 +495,7 @@ def config_json_for_scan(config: CrawlConfig) -> str:
"max_depth": config.max_depth,
"allow_subdomains": config.allow_subdomains,
"rps": config.rps,
"workers": config.workers,
"ignore_robots": config.ignore_robots,
"concurrency_per_host": config.concurrency_per_host,
"user_agent": config.user_agent,
Expand Down Expand Up @@ -784,9 +785,12 @@ async def _worker(ctx: _WorkerContext) -> None:
while True:
if _page_limit_reached(ctx):
return
job = queue.lease(ctx.conn, JOB_KIND, lease_secs=120)
job = queue.lease(ctx.conn, JOB_KIND, lease_secs=120, scan_id=ctx.scan_id)
if job is None:
if ctx.in_flight == 0 and queue.pending_count(ctx.conn, JOB_KIND) == 0:
if (
ctx.in_flight == 0
and queue.pending_count(ctx.conn, JOB_KIND, scan_id=ctx.scan_id) == 0
):
return
await asyncio.sleep(0.05)
continue
Expand Down
78 changes: 58 additions & 20 deletions src/audit/db/queue.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ def lease(
kind: str,
lease_secs: float,
*,
scan_id: int | None = None,
now: datetime | None = None,
) -> Job | None:
"""Atomically claim one pending job of ``kind``. Returns ``None`` when queue is empty.
Expand All @@ -73,23 +74,43 @@ def lease(
"""
current = _now(now)
lease_until = current + timedelta(seconds=lease_secs)
cur = conn.execute(
"""
UPDATE jobs
SET state = 'leased',
lease_until = ?,
attempts = attempts + 1,
updated_at = ?
WHERE id = (
SELECT id FROM jobs
WHERE state = 'pending' AND kind = ?
ORDER BY id
LIMIT 1
)
RETURNING id, kind, payload_json, attempts
""",
(_iso(lease_until), _iso(current), kind),
)
if scan_id is None:
cur = conn.execute(
"""
UPDATE jobs
SET state = 'leased',
lease_until = ?,
attempts = attempts + 1,
updated_at = ?
WHERE id = (
SELECT id FROM jobs
WHERE state = 'pending' AND kind = ?
ORDER BY id
LIMIT 1
)
RETURNING id, kind, payload_json, attempts
""",
(_iso(lease_until), _iso(current), kind),
)
else:
cur = conn.execute(
"""
UPDATE jobs
SET state = 'leased',
lease_until = ?,
attempts = attempts + 1,
updated_at = ?
WHERE id = (
SELECT id FROM jobs
WHERE state = 'pending' AND kind = ?
AND json_extract(payload_json, '$.scan_id') = ?
ORDER BY id
LIMIT 1
)
RETURNING id, kind, payload_json, attempts
""",
(_iso(lease_until), _iso(current), kind, scan_id),
)
row = cur.fetchone()
if row is None:
return None
Expand Down Expand Up @@ -153,14 +174,31 @@ def reclaim_expired(conn: sqlite3.Connection, *, now: datetime | None = None) ->
return int(cur.rowcount)


def pending_count(conn: sqlite3.Connection, kind: str | None = None) -> int:
def pending_count(
conn: sqlite3.Connection,
kind: str | None = None,
*,
scan_id: int | None = None,
) -> int:
"""Count jobs currently in ``pending`` state."""
if kind is None:
if kind is None and scan_id is None:
cur = conn.execute("SELECT COUNT(*) AS n FROM jobs WHERE state = 'pending'")
else:
elif kind is not None and scan_id is None:
cur = conn.execute(
"SELECT COUNT(*) AS n FROM jobs WHERE state = 'pending' AND kind = ?",
(kind,),
)
elif kind is None:
cur = conn.execute(
"SELECT COUNT(*) AS n FROM jobs WHERE state = 'pending' "
"AND json_extract(payload_json, '$.scan_id') = ?",
(scan_id,),
)
else:
cur = conn.execute(
"SELECT COUNT(*) AS n FROM jobs WHERE state = 'pending' AND kind = ? "
"AND json_extract(payload_json, '$.scan_id') = ?",
(kind, scan_id),
)
row = cur.fetchone()
return int(row["n"]) if row is not None else 0
37 changes: 33 additions & 4 deletions src/audit/protected/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -429,7 +429,34 @@ def verify_authenticated_target(self, url: str | None = None) -> ValidatedUrl:
self._state = ManualAuthState.AUTHENTICATED
return verified

async def minimize_for_background_scan(self) -> bool:
async def prepare_background_scan_pages(self, count: int) -> tuple[Page, ...]:
"""Create the fixed authenticated tab pool before hiding Chromium.

macOS restores a minimized Chromium window whenever Playwright creates
a new top-level page. Preparing every worker tab first lets the crawl
reuse those pages without repeatedly bringing the browser to the
foreground. The pages and their shared authenticated context remain
memory-only and are destroyed together at session close.
"""

if self._state is not ManualAuthState.AUTHENTICATED:
raise ManualAuthenticationError(
"Complete and verify manual sign-in before preparing scan tabs."
)
if count <= 0:
raise ValueError("Background scan page count must be positive.")
pages: list[Page] = []
try:
for _ in range(count):
pages.append(await self.context.new_page())
except Exception:
for page in pages:
with contextlib.suppress(Exception):
await page.close(run_before_unload=False)
raise
return tuple(pages)

async def minimize_for_background_scan(self, page: Page | None = None) -> bool:
"""Minimize the headed sign-in window before reusing its context.

Chromium cannot switch a live authenticated context from headed to
Expand All @@ -444,12 +471,12 @@ async def minimize_for_background_scan(self) -> bool:
raise ManualAuthenticationError(
"Complete and verify manual sign-in before backgrounding the browser."
)
page = self._page
if page is None:
target_page = page or self._page
if target_page is None:
return False
cdp_session: Any | None = None
try:
cdp_session = await self.context.new_cdp_session(page)
cdp_session = await self.context.new_cdp_session(target_page)
window = await cdp_session.send("Browser.getWindowForTarget")
window_id = window.get("windowId")
if not isinstance(window_id, int):
Expand Down Expand Up @@ -552,6 +579,7 @@ def create_shared_js_fetcher(
visual_probe: VisualProbe | None = None,
capture_screenshots: bool = False,
max_rendered_html_chars: int | None = None,
shared_pages: tuple[Page, ...] = (),
) -> JsFetcher:
"""Return a fetcher that reuses this in-memory authenticated context.

Expand All @@ -573,6 +601,7 @@ def create_shared_js_fetcher(
visual_probe=visual_probe,
capture_screenshots=capture_screenshots,
shared_context=self.context,
shared_pages=shared_pages,
private_context=True,
max_rendered_html_chars=max_rendered_html_chars,
)
Expand Down
2 changes: 2 additions & 0 deletions src/audit/web/frontend/src/api/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -312,6 +312,8 @@ export interface LocalLoginScanPayload {
max_pages: number;
max_depth: number;
rps: number;
/** Concurrent authenticated tabs; the local login API caps this at four. */
workers: number;
whole_host: boolean;
/** DOM rule engines run against the signed-in application scope. */
scan_engine: ProtectedScanEngine;
Expand Down
14 changes: 10 additions & 4 deletions src/audit/web/frontend/src/components/LocalLoginScan.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ function LocalLoginForm({ showSteps }: { showSteps: boolean }) {
max_pages: 2500,
max_depth: 10,
rps: 1,
workers: 2,
whole_host: false,
scan_engine: "axe",
axe_level: "AA",
Expand Down Expand Up @@ -278,6 +279,7 @@ function LocalLoginForm({ showSteps }: { showSteps: boolean }) {
? "Local OCR image-text analysis"
: "Local OCR and loopback VLM analysis",
`Up to ${form.max_pages.toLocaleString()} pages`,
`${form.workers} concurrent authenticated ${form.workers === 1 ? "tab" : "tabs"}`,
].map((label) => (
<li key={label} className="flex items-start gap-2">
<Check
Expand Down Expand Up @@ -321,13 +323,17 @@ function LocalLoginForm({ showSteps }: { showSteps: boolean }) {
/>
<LoginNumberField
label="Workers"
value={1}
value={form.workers}
min={1}
max={1}
disabled
onChange={() => undefined}
max={4}
onChange={(value) => update("workers", value)}
/>
</div>
<p className="text-xs text-fg-muted">
Workers open concurrent tabs inside the same temporary,
authenticated browser session. Two is recommended; four is
the safety maximum for login and 2FA scans.
</p>

<fieldset className="rounded-xs border border-border p-3">
<legend className="px-1 text-xs font-semibold uppercase tracking-wide text-fg-subtle">
Expand Down
Loading
Loading