diff --git a/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/agent.py b/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/agent.py new file mode 100644 index 0000000..15f15eb --- /dev/null +++ b/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/agent.py @@ -0,0 +1,1305 @@ +"""SN60 / Bitsec challenger agent -- a Solidity auditor built around storage def-use analysis. + +Most audit agents read one file, or one function, and ask a model "what is wrong here?". That +finds the bug classes visible locally -- a missing modifier, an unchecked return, a reentrant call +-- and systematically misses the ones that are not. A large share of real high-severity findings +are not local at all: one function mutates a storage accumulator, and some OTHER function reads it +later and depends on an invariant the first one quietly broke. Neither function is wrong on its +own. You only see it by reading both at once, and a file-at-a-time reader has no reason to put +those two functions in front of the model together. + +So this agent computes that pairing structurally, before spending a single token on it. + +It parses each contract's state variables and function bodies, builds a def-use graph over storage +-- which functions WRITE each variable, which READ it -- and ranks the resulting pairs by the +signals that make a broken invariant likely. The strongest signal is a KEY MISMATCH: the same +mapping written under one index expression and read under a different one. When a writer says +``totals[currentBucket] -= x`` and a reader says ``totals[previousBucket]``, the two are talking +about different buckets, and whether that is safe rests on an ordering assumption written down +nowhere. That is worth asking a model about, and it is a question the model cannot be asked at all +unless something first noticed those two functions were related. + +The pairing is expensive to discover and cheap to check, so structure proposes and inference +decides. Nothing here encodes a known bug, a project, or an expected answer: the analysis is +ordinary def-use over whatever contracts it is handed, every candidate is offered to the model as a +suspicion that may well be nothing, and every verdict comes from inference over the real source. + +Pipeline: + 1. catalog -- find contracts, rank by reachable attack surface + 2. parse -- functions, state variables, visibility, guards + 3. def-use -- storage write/read graph; rank cross-function pairs by signal + 4. probe -- targeted prompts: one storage variable, its writers and readers, side by side + 5. survey -- a project-wide map, for what is visible from shape alone + 6. focus -- deep per-file audit of the highest-surface contracts + 7. select -- normalise, dedupe, corroborate, rank, cap + +Runtime contract (enforced by screening and the sandbox runner): + * ``agent_main()`` is synchronous and callable with no arguments + * it returns ``{"vulnerabilities": [...]}`` + * ``PROJECT_DIR`` (or ``/app/project_code``) holds the code under audit + * ``INFERENCE_API`` + ``INFERENCE_API_KEY`` are the miner-funded in-room endpoint +""" + +from __future__ import annotations + +import json +import os +import re +import socket +import threading +import time +import urllib.error +import urllib.request +from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path + + +# --------------------------------------------------------------------------------------------- +# Configuration. Every value is env-overridable so the agent can be retuned without a code change. +# --------------------------------------------------------------------------------------------- + +def _num(name: str, default: float) -> float: + try: + return float(os.environ.get(name, "").strip() or default) + except ValueError: + return default + + +def _int(name: str, default: int) -> int: + return int(_num(name, default)) + + +#: Chosen for two properties that turned out to be independent: it reaches a VERDICT instead of +#: enumerating scenarios indefinitely, and it emits the JSON it is asked for. Verbose reasoning +#: models measured alongside it analysed capably but spent their whole token budget narrating and +#: were truncated before writing any conclusion, which scores exactly the same as finding nothing. +#: A model that concludes briefly beats one that reasons longer and never answers. +MODEL = os.environ.get("AGENT_MODEL", "deepseek-ai/DeepSeek-V4-Flash").strip() + +#: The whole run must finish INSIDE the sealed room's request lifetime, and that is a hard wall +#: rather than a target. The validator POSTs ``/run`` once and blocks on that single socket; the +#: room stamps the request ``expires_at = issued_at + lifetime`` (900s by default, 1200s ceiling) +#: and the caller's own socket timeout matches it. An agent still working when that clock runs out +#: does not return a partial report -- the socket times out, the transport layer treats it as a +#: room that may never have run, and it RETRIES with a fresh nonce, up to three times. Each retry +#: re-runs the agent from scratch and spends the miner's inference budget again on a report that +#: can never be delivered. So overshooting the budget is not "a slower run", it is three paid runs +#: and no result. The margin below covers the room's own overhead: untarring the bundle, fetching +#: and extracting the project, starting the agent container, and shipping the report back. +#: Sized from what the passes actually YIELD, not from what the room allows. Every finding this +#: agent has produced came from the structural probes, which run first and finish early; the broad +#: survey/focus passes that consumed the rest of the budget contributed none. Burning the room's +#: full clock re-reading files that yield nothing costs the miner money and makes every screening +#: gate and every scored replica slower for no gain. +WALL_BUDGET = _num("AGENT_WALL_BUDGET", 420.0) +TAIL_RESERVE = _num("AGENT_TAIL_RESERVE", 30.0) +PER_CALL_TIMEOUT = _num("AGENT_CALL_TIMEOUT", 150.0) +CALL_FLOOR = _num("AGENT_CALL_FLOOR", 20.0) +CALL_RETRIES = _int("AGENT_CALL_RETRIES", 2) +#: Concurrency buys breadth inside a fixed wall clock, but only up to the point where the PROVIDER +#: starts queueing. Measured against the in-room gateway's upstream: at 12 in flight, 9 of 26 calls +#: died of timeouts and dropped connections, because the queue delay pushed each one past its own +#: deadline. Past that knee, more concurrency returns FEWER completed calls, not more -- so this +#: sits deliberately below it. The work is network-bound, so the local cost of either setting is nil; +#: the only thing being traded is how much the upstream is willing to serve at once. +WORKERS = max(1, _int("AGENT_WORKERS", 7)) +DECODE_TEMP = _num("AGENT_TEMPERATURE", 0.15) + +MAX_FILES = _int("AGENT_MAX_FILES", 90) +FOCUS_FILES = _int("AGENT_FOCUS_FILES", 3) +PROBE_LIMIT = _int("AGENT_PROBE_LIMIT", 10) +DESYNC_LIMIT = _int("AGENT_DESYNC_LIMIT", 6) +MAX_FINDINGS = _int("AGENT_MAX_FINDINGS", 40) + +#: A confirmed finding measured ~400 tokens. The old 3000 ceiling bought no better analysis: +#: generation latency scales with tokens PRODUCED, so an oversized cap only let a verbose model +#: ramble for minutes and, at worst, get truncated before writing its JSON at all. +PROBE_TOKENS = _int("AGENT_PROBE_TOKENS", 1800) +FOCUS_TOKENS = _int("AGENT_FOCUS_TOKENS", 2000) +SURVEY_TOKENS = _int("AGENT_SURVEY_TOKENS", 2200) +MIN_TOKENS = _int("AGENT_MIN_TOKENS", 700) +SALVAGE_TOKENS = _int("AGENT_SALVAGE_TOKENS", 1400) +SALVAGE_CHARS = _int("AGENT_SALVAGE_CHARS", 14000) + +FOCUS_CHARS = _int("AGENT_FOCUS_CHARS", 46000) +PROBE_CHARS = _int("AGENT_PROBE_CHARS", 30000) +SURVEY_CHARS = _int("AGENT_SURVEY_CHARS", 42000) +FN_CHARS = _int("AGENT_FN_CHARS", 5000) + +RETRYABLE = frozenset({408, 409, 425, 429, 500, 502, 503, 504}) +SKIP_DIRS = ("test", "tests", "mock", "mocks", "script", "scripts", "node_modules", + "artifacts", "cache", "docs") + +# Endpoint capability probing. The in-room gateway fronts several providers and they do not accept +# the same optional parameters; a 400 on one of these is a dialect difference, not a failure, so +# each is disabled once and globally rather than rediscovered on every call. +_PARAM_LOCK = threading.Lock() +_EFFORT_OK = True +_TEMP_OK = True +_JSON_MODE_OK = True +_TOKEN_CEILING = 0 + +SYSTEM = ( + "You are a senior smart-contract security auditor. You report only vulnerabilities you can " + "justify from the code in front of you, and you name the exact file, contract and function " + "where each one lives. You do not speculate and you do not pad." +) + +#: The finding shape, identical across every pass so results from different passes are directly +#: comparable and can be deduped against one another. +SHAPE = ( + '{"findings":[{"title":str,"severity":"high"|"critical","type":str,"file":str,' + '"contract":str,"function":str,"description":str,"confidence":0.0-1.0}]}' +) + +RUBRIC = ( + "Report only HIGH or CRITICAL severity issues: a way to steal or lock funds, to corrupt " + "accounting so balances or rewards come out wrong, to bypass authorization, or to brick a core " + "flow. Ignore gas, style, naming, events, and unreachable or test-only code.\n" + "Each description MUST state, concretely:\n" + " 1. the file, by its .sol name, and the function it lives in, written as functionName();\n" + " 2. the mechanism -- which state or value ends up wrong, and the exact code that makes it so;\n" + " 3. why the checks already present do not prevent it;\n" + " 4. the consequence, in terms of funds or accounting.\n" + "Write EVERY function you mention as functionName(), with the parentheses -- including the " + "other functions that participate in the issue, not only the one it is filed under. An issue " + "that spans two functions is only reproducible if both are named, and the state variable " + "involved should be named too.\n" + "If a check already present stops the issue, do not report it." +) + + +# --------------------------------------------------------------------------------------------- +# Inference +# --------------------------------------------------------------------------------------------- + +def endpoint_url() -> str: + return (os.environ.get("INFERENCE_API") or "").strip().rstrip("/") + + +def _payload(prompt: str, max_tokens: int, effort: str | None, with_temp: bool, + json_mode: bool) -> bytes: + body: dict[str, object] = { + "model": MODEL, + "messages": [ + {"role": "system", "content": SYSTEM}, + {"role": "user", "content": prompt}, + ], + "max_tokens": max_tokens, + } + if with_temp: + body["temperature"] = DECODE_TEMP + if effort: + body["reasoning_effort"] = effort + # Asking for JSON in the prompt is a request; ``response_format`` is a constraint. Instruction + # alone does not hold: a model with a strong chain-of-thought habit will narrate until it hits + # the token ceiling and never reach the closing brace, so a correct analysis arrives as prose + # and parses to nothing. Constraining the decoder is what makes the answer retrievable. + if json_mode: + body["response_format"] = {"type": "json_object"} + return json.dumps(body).encode("utf-8") + + +def _message_text(payload: object) -> str: + """Pull the assistant text out of a chat completion. + + Reasoning models sometimes leave ``content`` empty and put the substance in a reasoning field. + Treating that as an empty answer throws away a call that has already been paid for. + """ + if not isinstance(payload, dict): + return "" + choices = payload.get("choices") + if not isinstance(choices, list) or not choices or not isinstance(choices[0], dict): + return "" + message = choices[0].get("message") + if not isinstance(message, dict): + return "" + content = message.get("content") + if isinstance(content, list): + content = "".join(p.get("text", "") for p in content if isinstance(p, dict)) + if isinstance(content, str) and content.strip(): + return content + for alternate in ("reasoning", "reasoning_content"): + value = message.get(alternate) + if isinstance(value, str) and value.strip(): + return value + details = message.get("reasoning_details") + if isinstance(details, list): + joined = "".join(p.get("text", "") for p in details if isinstance(p, dict)) + if joined.strip(): + return joined + return "" + + +def ask(endpoint: str, prompt: str, deadline: float, max_tokens: int, + effort: str | None = "medium", json_mode: bool = True) -> str: + global _EFFORT_OK, _TEMP_OK, _JSON_MODE_OK, _TOKEN_CEILING + if not endpoint: + raise RuntimeError("no inference endpoint") + headers = { + "Content-Type": "application/json", + "x-inference-api-key": os.environ.get("INFERENCE_API_KEY", ""), + } + last: object = None + backoff = 2.0 + attempt = 0 + while attempt < CALL_RETRIES: + remaining = deadline - time.monotonic() - TAIL_RESERVE + timeout = min(PER_CALL_TIMEOUT, float(int(remaining))) + if timeout < CALL_FLOOR: + raise RuntimeError("time budget exhausted") + use_effort = effort if _EFFORT_OK else None + use_temp = _TEMP_OK + use_json = json_mode and _JSON_MODE_OK + want = min(max_tokens, _TOKEN_CEILING) if _TOKEN_CEILING else max_tokens + request = urllib.request.Request( + endpoint + "/inference", + data=_payload(prompt, want, use_effort, use_temp, use_json), + method="POST", + headers=headers, + ) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + return _message_text(json.loads(response.read().decode("utf-8", "replace"))) + except urllib.error.HTTPError as exc: + # A 400 here is the endpoint rejecting an optional parameter, not a bad request. Drop + # the parameter and retry WITHOUT consuming an attempt: the request itself was fine. + if exc.code == 400 and use_effort: + with _PARAM_LOCK: + _EFFORT_OK = False + continue + if exc.code == 400 and use_temp: + with _PARAM_LOCK: + _TEMP_OK = False + continue + if exc.code == 400 and use_json: + with _PARAM_LOCK: + _JSON_MODE_OK = False + continue + if exc.code == 400 and want > MIN_TOKENS: + with _PARAM_LOCK: + _TOKEN_CEILING = max(MIN_TOKENS, want // 2) + continue + if exc.code not in RETRYABLE: + raise RuntimeError("http %d" % exc.code) from exc + last = exc + except (socket.timeout, TimeoutError) as exc: + raise RuntimeError("call timed out") from exc + except urllib.error.URLError as exc: + if isinstance(exc.reason, (socket.timeout, TimeoutError)): + raise RuntimeError("call timed out") from exc + last = exc + except (OSError, ValueError) as exc: + last = exc + attempt += 1 + if attempt >= CALL_RETRIES: + break + if deadline - time.monotonic() <= backoff + TAIL_RESERVE + CALL_FLOOR: + break + time.sleep(backoff) + backoff = min(backoff * 2.0, 8.0) + raise RuntimeError(str(last) if last else "request failed") + + +# --------------------------------------------------------------------------------------------- +# Solidity parsing. Deliberately lightweight -- brace matching and regex, no grammar. It has to be +# robust to whatever compiles, not correct for every pathological input: a missed function costs +# one probe, whereas a parser that raises costs the entire run. +# --------------------------------------------------------------------------------------------- + +_LINE_COMMENT = re.compile(r"//[^\n]*") +_BLOCK_COMMENT = re.compile(r"/\*.*?\*/", re.S) +_FUNCTION = re.compile(r"\bfunction\s+(\w+)\s*\(([^)]*)\)\s*([^{;]*?)(\{|;)", re.S) +_CONTRACT = re.compile(r"\b(?:contract|library|interface|abstract\s+contract)\s+(\w+)") +_STATE_MAPPING = re.compile( + r"\bmapping\s*\(([^;{}]*?)\)\s*(?:public|private|internal)?\s*(\w+)\s*;") +_STATE_SCALAR = re.compile( + r"^[ \t]*(u?int\d*|address|bytes\d*|bool)\s+(?:public|private|internal)?\s*" + r"(?:immutable\s+)?(\w+)\s*(?:=[^;]*)?;", re.M) +_GUARD_WORDS = ("only", "auth", "require(", "revert", "assert(", "nonreentrant", "whennotpaused") + + +def strip_comments(text: str) -> str: + return _LINE_COMMENT.sub("", _BLOCK_COMMENT.sub("", text)) + + +def match_block(text: str, brace_index: int) -> int: + """Index just past the block opened at ``brace_index``, or -1 if unbalanced.""" + depth = 0 + for i in range(brace_index, len(text)): + char = text[i] + if char == "{": + depth += 1 + elif char == "}": + depth -= 1 + if depth == 0: + return i + 1 + return -1 + + +class Function: + __slots__ = ("name", "params", "attrs", "body", "start", "end", "line") + + def __init__(self, name, params, attrs, body, start, end, line): + self.name = name + self.params = params + self.attrs = attrs + self.body = body + self.start = start + self.end = end + self.line = line + + @property + def external(self) -> bool: + return "external" in self.attrs or "public" in self.attrs + + @property + def guarded(self) -> bool: + blob = (self.attrs + " " + self.body[:400]).lower() + return any(word in blob for word in _GUARD_WORDS) + + def signature(self) -> str: + flags = " ".join(w for w in self.attrs.split() if w in ( + "external", "public", "internal", "private", "view", "pure", "payable")) + return "%s(%s) %s" % (self.name, " ".join(self.params.split())[:90], flags) + + +def parse_functions(clean: str) -> list[Function]: + out: list[Function] = [] + for match in _FUNCTION.finditer(clean): + if match.group(4) == ";": # interface / abstract declaration: no body to analyse + continue + brace = clean.find("{", match.end() - 1) + if brace < 0: + continue + end = match_block(clean, brace) + if end < 0: + continue + out.append(Function( + name=match.group(1), + params=match.group(2), + attrs=" ".join(match.group(3).split()), + body=clean[brace:end], + start=match.start(), + end=end, + line=clean.count("\n", 0, match.start()) + 1, + )) + return out + + +def parse_state(clean: str, functions: list[Function]) -> dict[str, str]: + """State variables, as ``{name: declaration}``. + + Function bodies are cut out before scanning, because anything declared inside one is a local. + Without that every local ``uint256 amount`` looks like a storage slot and the def-use graph + fills up with noise that crowds out the real accumulators. + """ + kept: list[str] = [] + cursor = 0 + for fn in sorted(functions, key=lambda f: f.start): + if fn.start > cursor: + kept.append(clean[cursor:fn.start]) + cursor = max(cursor, fn.end) + kept.append(clean[cursor:]) + scope = "\n".join(kept) + + state: dict[str, str] = {} + for match in _STATE_MAPPING.finditer(scope): + state[match.group(2)] = "mapping(%s) %s" % ( + " ".join(match.group(1).split()), match.group(2)) + for match in _STATE_SCALAR.finditer(scope): + name = match.group(2) + # SCREAMING_CASE is conventionally a constant, and constants cannot break an invariant. + if name not in state and name.upper() != name: + state[name] = "%s %s" % (match.group(1), name) + return state + + +# --------------------------------------------------------------------------------------------- +# Storage def-use. The core of this agent. +# --------------------------------------------------------------------------------------------- + +_ASSIGN = r"(?:=(?!=)|\+=|-=|\*=|/=|\|=|&=|\^=|\+\+|--)" + + +def _write_re(var: str) -> re.Pattern: + name = re.escape(var) + return re.compile( + r"\bdelete\s+%s\b" # delete v, delete v[k] + r"|\b%s\s*(?:\[[^\]]*\]\s*)*(?:\.\w+\s*)*%s" % (name, name, _ASSIGN)) + + +def _keys(body: str, var: str) -> set[str]: + """Index expressions this body applies to ``var``, whitespace-normalised.""" + return {" ".join(raw.split()) + for raw in re.findall(re.escape(var) + r"\s*\[([^\[\]]{1,90})\]", body)} + + +class Usage: + __slots__ = ("var", "decl", "writers", "readers") + + def __init__(self, var: str, decl: str): + self.var = var + self.decl = decl + self.writers: dict[str, set[str]] = {} + self.readers: dict[str, set[str]] = {} + + +def storage_usage(functions: list[Function], state: dict[str, str]) -> list[Usage]: + """For each state variable, which functions write it and which only read it.""" + usages = [] + for var, decl in state.items(): + writer = _write_re(var) + reader = re.compile(r"\b%s\b" % re.escape(var)) + usage = Usage(var, decl) + for fn in functions: + if not reader.search(fn.body): + continue + keys = _keys(fn.body, var) + if writer.search(fn.body): + usage.writers[fn.name] = keys + else: + usage.readers[fn.name] = keys + # Only cross-function pairs are interesting: a variable written and read by the same + # function alone has no second party whose assumption could be violated. + if usage.writers and usage.readers: + usages.append(usage) + return usages + + +#: Words marking an index as a time bucket. A key mismatch matters far more when the index is +#: temporal, because then the two functions disagree about WHEN, and the ordering assumption that +#: reconciles them is almost never written down. Generic DeFi vocabulary, not any project's. +_TEMPORAL = ("epoch", "period", "week", "day", "time", "cycle", "round", + "checkpoint", "snapshot", "block", "term", "phase", "start", "last") + +#: Names that suggest a running total. A wrong write to one of these is not self-correcting: it is +#: carried forward into every later read. +_ACCUMULATOR = ("total", "sum", "index", "acc", "supply", "reserve", "weight", "balance", + "debt", "share", "reward", "rate", "cumul", "liquidity", "stake") + + +def score_usage(usage: Usage) -> tuple[int, list[str]]: + """Rank a storage variable by how likely a writer is to break a reader's assumption. + + Signals are weighted by how specific they are. The key mismatch dominates because it is a + concrete disagreement in the code rather than a smell: two functions index the same mapping + differently, so at most one of them can be right about which bucket holds the truth. + """ + reasons: list[str] = [] + score = 0 + + write_keys: set[str] = set() + read_keys: set[str] = set() + for keys in usage.writers.values(): + write_keys |= keys + for keys in usage.readers.values(): + read_keys |= keys + + if write_keys and read_keys and (write_keys ^ read_keys): + score += 50 + reasons.append("written and read under different index expressions") + combined = " ".join(write_keys | read_keys).lower() + temporal = any(word in combined for word in _TEMPORAL) + if temporal: + score += 25 + reasons.append("the differing index is a time bucket, so the two disagree about WHEN") + + write_offset = any(re.search(r"[+\-]", key) for key in write_keys) + read_offset = any(re.search(r"[+\-]", key) for key in read_keys) + if read_offset != write_offset: + # One side is looking at a SHIFTED bucket. That is the case worth spelling out, + # because the succession is what makes it a bug and the succession is invisible when + # you read either function on its own: the bucket a writer mutates "now" is the very + # bucket the offset reader will consult one period later. A reviewer -- human or model + # -- who assumes past periods are frozen will wave this through, since nothing in + # either function says otherwise. + score += 20 + if read_offset: + reasons.append( + "the reader consults a SHIFTED (earlier) bucket while writers use the " + "unshifted (current) one. So the bucket writers mutate throughout period P is " + "exactly the bucket the reader consumes in period P+1 to account for P. Every " + "write during P is therefore retroactive from the reader's point of view: it " + "is applied while P is still open, but it changes the number P is ultimately " + "settled against. The question is NOT whether a past bucket can be edited " + "(it usually cannot) -- it is whether each write made DURING P is still " + "correct once that bucket is used as P's total") + else: + reasons.append( + "writers use a SHIFTED bucket while the reader uses the unshifted one, so a " + "write can land in a period the reader has already accounted for") + elif read_offset and write_offset: + score += 10 + reasons.append("both sides offset their index, so the shift has to agree exactly") + + low = usage.var.lower() + if any(word in low for word in _ACCUMULATOR): + score += 12 + reasons.append("the variable is a running total, so a wrong write is carried forward") + + if len(usage.readers) >= 2: + score += 6 + reasons.append("several functions depend on it") + if "mapping" in usage.decl: + score += 4 + return score, reasons + + +#: Prefixes that mark a variable as the AGGREGATE of another: ``totalShares`` is the sum over +#: holders of ``shares``. The pair is discovered by stripping one of these off the name. +_AGGREGATE_PREFIXES = ("total", "sum", "aggregate", "global", "cumulative", "accumulated", "all") + + +def aggregate_pairs(state: dict[str, str]) -> list[tuple[str, str]]: + """``(aggregate, component)`` pairs, found by name. + + A contract that keeps both a running total and its per-key parts has an invariant it never + writes down: the total equals the sum of the parts. Every function that changes one must change + the other, and the pair is only discoverable by looking at the names. + """ + pairs = [] + lowered = {name.lower(): name for name in state} + for name in state: + low = name.lower() + for prefix in _AGGREGATE_PREFIXES: + if not low.startswith(prefix) or len(low) <= len(prefix): + continue + stem = low[len(prefix):] + for candidate_low, candidate in lowered.items(): + if candidate_low == stem and candidate != name: + pairs.append((name, candidate)) + break + return pairs + + +def desync_findings(functions: list[Function], state: dict[str, str]) -> list[dict]: + """Functions that mutate an aggregate without its component, or vice versa. + + This is the shape of a broken sum. When most functions update both sides and ONE updates only + one of them, that one is the anomaly -- and the majority is what makes it an anomaly rather + than a design choice, which is why the other functions are reported alongside it as the + contrast. The imbalance survives in storage and every later reader of the total inherits it. + + Deliberately name-driven and therefore approximate: it proposes a question, and the model + decides. A pair that is genuinely unrelated simply produces a probe that comes back empty. + """ + out = [] + for aggregate, component in aggregate_pairs(state): + agg_writer = _write_re(aggregate) + comp_writer = _write_re(component) + both, only_aggregate, only_component = [], [], [] + for fn in functions: + writes_aggregate = bool(agg_writer.search(fn.body)) + writes_component = bool(comp_writer.search(fn.body)) + if writes_aggregate and writes_component: + both.append(fn.name) + elif writes_aggregate: + only_aggregate.append(fn.name) + elif writes_component: + only_component.append(fn.name) + # Without a function that maintains both, there is no established convention to deviate + # from, and "updates only one" is just how this contract is written. + if not both or not (only_aggregate or only_component): + continue + out.append({ + "aggregate": aggregate, + "component": component, + "both": both, + "only_aggregate": only_aggregate, + "only_component": only_component, + }) + return out + + +def desync_prompt(rec: dict, desync: dict) -> str: + aggregate, component = desync["aggregate"], desync["component"] + head = ( + "This contract keeps a running total and the per-key parts it is supposed to be the sum of. " + "Some functions update both and keep them consistent. At least one updates only ONE side.\n\n" + "Decide whether that asymmetry breaks the invariant `%s == sum of %s`, and if it does, what " + "goes wrong for the code that later reads either value.\n\n" + "Check specifically:\n" + " 1. Does the odd function leave the total disagreeing with the parts? Follow the actual " + "arithmetic -- a subtraction from the total that does not clear the corresponding part " + "leaves that part still counted by anyone who iterates the parts.\n" + " 2. Is there an inverse operation elsewhere (a revive, a re-enable, an undo) that " + "restores one side but not the other? An asymmetric pair is a common way this breaks.\n" + " 3. Who consumes these values afterwards -- especially anything that DIVIDES by the " + "total or pays out per part. If the total is smaller than the sum of the parts, a payout " + "computed per part against a rate derived from the total pays out more in aggregate than " + "was ever provided.\n\n" + 'If the asymmetry is harmless, say so and return {"findings":[]}. Reach a verdict; do not ' + "keep enumerating scenarios.\n\n" + "FILE: %s\n" + "AGGREGATE: %s\nCOMPONENT: %s\n" + "functions updating BOTH (the intended convention): %s\n" + "functions updating ONLY the aggregate: %s\n" + "functions updating ONLY the component: %s\n\n" + % (aggregate, component, rec["rel"], state_decl(rec, aggregate), state_decl(rec, component), + ", ".join("%s()" % n for n in desync["both"]) or "none", + ", ".join("%s()" % n for n in desync["only_aggregate"]) or "none", + ", ".join("%s()" % n for n in desync["only_component"]) or "none") + + RUBRIC + "\nReturn strict JSON only: " + SHAPE + "\n" + ) + names = desync["only_aggregate"] + desync["only_component"] + desync["both"] + # Any function that reads either side is a potential victim, and the payout/divide site is + # usually one of them -- without it the model cannot see what the imbalance costs. + readers = [f.name for f in rec["functions"] + if f.name not in names + and (re.search(r"\b%s\b" % re.escape(aggregate), f.body) + or re.search(r"\b%s\b" % re.escape(component), f.body))] + room = PROBE_CHARS - len(head) + ordered = names + readers[:3] + share = max(600, room // max(1, len(ordered))) + blocks = [] + for name in ordered: + text = function_text(rec, name, share) + if not text: + continue + block = "\n===== %s() =====\n%s\n" % (name, text) + if len(block) > room: + break + blocks.append(block) + room -= len(block) + return head + "".join(blocks) + + +def state_decl(rec: dict, name: str) -> str: + return rec["state"].get(name, name) + + +def build_probes(records: list[dict]) -> list[dict]: + """Cross-function storage probes across the project, strongest signal first.""" + probes = [] + for rec in records: + for usage in rec["usage"]: + score, reasons = score_usage(usage) + if score < 20: + continue + # The file's rank is only a TIEBREAKER, and is capped hard. It measures attack surface + # for the whole file, which is the right prior for deciding what to read deeply but the + # wrong one for ranking a specific variable: a big, busy contract earns a large rank + # from sheer size and would otherwise outrank a precise def-use signal found in a + # smaller one. Letting it in unbounded buried the strongest signals in this project + # under routine candidates from the largest file. + probes.append({ + "rel": rec["rel"], + "usage": usage, + "reasons": reasons, + "rec": rec, + "score": score + min(10, max(0, rec["rank"]) // 30), + }) + probes.sort(key=lambda p: -p["score"]) + return probes[:PROBE_LIMIT] + + +def function_text(rec: dict, name: str, limit: int) -> str: + for fn in rec["functions"]: + if fn.name == name: + body = fn.body + if len(body) > limit: + body = body[:limit] + "\n/* ... truncated ... */" + return "function %s(%s) %s %s" % (fn.name, fn.params, fn.attrs, body) + return "" + + +def probe_prompt(probe: dict) -> str: + usage = probe["usage"] + rec = probe["rec"] + head = "".join([ + "A structural pass over this contract found one storage variable written by some functions " + "and read by others, with the signals listed below. Decide whether any writer can leave " + "that variable in a state which makes a reader compute a WRONG result.\n\n" + "Work through it in this order, briefly:\n" + " 1. For each reader, state the invariant it needs -- what must be true of this variable " + "at the moment it reads, for its result to be correct.\n" + " 2. Pick ONE concrete bucket and follow it through time. Call the period it belongs to " + "P. List every write that lands in that bucket while P is open, and then say exactly what " + "the reader computes from it afterwards.\n" + " 3. Now the decisive question: is every one of those writes still correct once that " + "bucket is used for what the reader uses it for? A write can be perfectly reasonable as a " + "book-keeping update for the live period and still be wrong as part of the total that " + "period is finally settled against -- especially a subtraction, which removes a " + "contribution that was counted while entitlements for P were accruing.\n" + " 4. Give a verdict.\n\n" + "Do not spend your budget proving a bucket cannot be edited after its period ends; assume " + "it cannot. The interesting case is the writes that happen while the period is still open. " + "Once you have a verdict, stop and write the JSON.\n\n" + "This is a CANDIDATE, not a known bug, and most candidates are fine. If every writer " + 'genuinely preserves what the readers need, return {"findings":[]}. Report only what you ' + "can trace in the code below.\n\n", + "FILE: %s\n" % probe["rel"], + "STORAGE: %s\n" % usage.decl, + "SIGNALS: %s\n" % "; ".join(probe["reasons"]), + "WRITERS: %s\n" % ", ".join( + "%s() indexes [%s]" % (n, " | ".join(sorted(k)) if k else "no index") + for n, k in usage.writers.items()), + "READERS: %s\n" % ", ".join( + "%s() indexes [%s]" % (n, " | ".join(sorted(k)) if k else "no index") + for n, k in usage.readers.items()), + "\n" + RUBRIC + "\nReturn strict JSON only: " + SHAPE + "\n", + ]) + + room = PROBE_CHARS - len(head) + names = list(usage.writers) + list(usage.readers) + share = max(600, room // max(1, len(names))) + blocks = [] + for name in names: + text = function_text(rec, name, share) + if not text: + continue + role = "WRITER" if name in usage.writers else "READER" + block = "\n===== %s: %s() =====\n%s\n" % (role, name, text) + if len(block) > room: + break + blocks.append(block) + room -= len(block) + return head + "".join(blocks) + + +# --------------------------------------------------------------------------------------------- +# Cataloguing +# --------------------------------------------------------------------------------------------- + +#: Contract languages this agent can read. The benchmark is NOT Solidity-only -- several projects +#: are written in Vyper, and one of them is what the pre-scoring execution screener runs candidates +#: against. An agent that globs ``*.sol`` alone finds almost nothing there, returns an empty report +#: in about a second, and is failed by the screener before it ever reaches a project it could score. +SOURCE_SUFFIXES = (".sol", ".vy") + +_VY_DEF = re.compile(r"^def\s+(\w+)\s*\(", re.M) +_VY_DECORATOR = re.compile(r"^@(\w+)", re.M) +#: Module-level ``name: type`` -- Vyper's storage declaration. Anchored at column 0 so locals, +#: which are always inside an indented body, cannot be mistaken for storage. +_VY_STATE = re.compile(r"^([a-zA-Z_]\w*)\s*:\s*(?!=)([^\n=]+)$", re.M) +_VY_NOT_STATE = frozenset({"implements", "event", "interface", "struct", "enum", "flag", "from", + "import", "def", "class", "if", "else", "for", "while", "return"}) + + +def parse_functions_vyper(clean: str) -> list[Function]: + """Vyper functions, delimited by indentation rather than braces. + + A body runs from the ``def`` line until the next line at column 0 that is not blank and not a + decorator, which is the same rule the language itself uses. + """ + lines = clean.splitlines(keepends=True) + starts: list[tuple[int, str, int]] = [] + offset = 0 + offsets = [] + for line in lines: + offsets.append(offset) + offset += len(line) + for index, line in enumerate(lines): + match = _VY_DEF.match(line) + if match: + starts.append((index, match.group(1), offsets[index])) + + out: list[Function] = [] + for position, (index, name, char_start) in enumerate(starts): + end_line = len(lines) + for probe in range(index + 1, len(lines)): + text = lines[probe] + if not text.strip() or text[:1].isspace(): + continue + if text.startswith("@"): # decorators belong to the NEXT function + end_line = probe + break + end_line = probe + break + body = "".join(lines[index:end_line]) + # Decorators sit ABOVE the def, so visibility is found by looking backwards. + attrs = [] + for back in range(index - 1, -1, -1): + stripped = lines[back].strip() + if not stripped: + continue + if stripped.startswith("@"): + attrs.append(stripped[1:].split("(")[0]) + continue + break + signature = lines[index] + params = signature[signature.find("(") + 1:signature.rfind(")")] if "(" in signature else "" + out.append(Function( + name=name, + params=params, + attrs=" ".join(reversed(attrs)), + body=body, + start=char_start, + end=char_start + len(body), + line=index + 1, + )) + return out + + +def parse_state_vyper(clean: str, functions: list[Function]) -> dict[str, str]: + state: dict[str, str] = {} + for match in _VY_STATE.finditer(clean): + name, declared = match.group(1), match.group(2).strip() + if name in _VY_NOT_STATE or name.upper() == name: + continue + if declared.endswith(":") or declared.startswith("#"): + continue + state[name] = "%s: %s" % (name, declared[:80]) + return state + + +def parse_source(clean: str, suffix: str) -> tuple[list[Function], dict[str, str]]: + if suffix == ".vy": + functions = parse_functions_vyper(clean) + return functions, parse_state_vyper(clean, functions) + functions = parse_functions(clean) + return functions, parse_state(clean, functions) + + +def find_root(project_dir: str | None) -> Path | None: + for candidate in (project_dir, os.environ.get("PROJECT_DIR"), "/app/project_code", "."): + if not candidate: + continue + path = Path(candidate).expanduser() + if path.is_dir() and any( + any(path.rglob("*" + suffix)) for suffix in SOURCE_SUFFIXES): + return path + return None + + +def rank_file(rel: str, text: str, functions: list[Function]) -> int: + """How much reachable attack surface a file has. Drives which files get a deep read.""" + low = (rel + " " + text).lower() + rank = 14 * sum(1 for f in functions if f.external and not f.guarded) + rank += 3 * sum(1 for f in functions if f.external) + for word, weight in ( + ("transfer", 8), ("call{", 10), ("delegatecall", 14), ("mint", 8), ("burn", 8), + ("withdraw", 10), ("deposit", 8), ("claim", 8), ("distribute", 9), ("reward", 7), + ("swap", 7), ("liquidat", 9), ("collateral", 8), ("oracle", 8), ("price", 6), + ("vote", 6), ("stake", 6), ("upgrade", 7), ("initialize", 6), ("permit", 6), + ): + if word in low: + rank += weight + for word, penalty in (("mock", 60), ("test", 50), ("example", 30), ("interface", 25)): + if word in rel.lower(): + rank -= penalty + return rank + + +def catalog(root: Path) -> list[dict]: + records = [] + candidates = [] + for suffix in SOURCE_SUFFIXES: + candidates.extend(root.rglob("*" + suffix)) + for path in sorted(candidates): + rel = path.relative_to(root).as_posix() + parts = {p.lower() for p in Path(rel).parts[:-1]} + if parts & set(SKIP_DIRS): + continue + try: + raw = path.read_text(encoding="utf-8", errors="replace") + except OSError: + continue + if not raw.strip() or len(raw) > 900_000: + continue + suffix = path.suffix.lower() + # Vyper comments are ``#`` and its docstrings are triple-quoted, so the Solidity comment + # stripper would leave them intact -- harmless, since it only ever removes text. + clean = strip_comments(raw) if suffix == ".sol" else raw + functions, state = parse_source(clean, suffix) + if not functions: + continue + records.append({ + "rel": rel, + "text": raw, + "functions": functions, + "state": state, + "contracts": _CONTRACT.findall(clean), + "usage": storage_usage(functions, state), + "rank": rank_file(rel, raw, functions), + }) + records.sort(key=lambda r: -r["rank"]) + return records[:MAX_FILES] + + +def condense(text: str, limit: int) -> str: + """Keep the head and tail of an over-long file; the middle is usually the least load-bearing.""" + if len(text) <= limit: + return text + head = int(limit * 0.7) + return text[:head] + "\n/* ... elided ... */\n" + text[-(limit - head):] + + +def survey_prompt(records: list[dict]) -> str: + head = ( + "Below is a structural map of a smart-contract project: per file its contracts, its " + "function signatures, and its storage variables. Report every high or critical issue " + "already justifiable from this shape alone -- unguarded external mutators, privileged " + "operations with no access control, initialization that can be front-run, accounting that " + "cannot hold.\n\n" + RUBRIC + "\nReturn strict JSON only: " + SHAPE + "\n\nPROJECT MAP:\n" + ) + parts = [head] + room = SURVEY_CHARS - len(head) + for rec in records: + block = ["\n## %s" % rec["rel"]] + if rec["contracts"]: + block.append("contracts: " + ", ".join(rec["contracts"][:6])) + if rec["state"]: + block.append("storage: " + ", ".join(list(rec["state"])[:14])) + externals = [f.signature() for f in rec["functions"] if f.external][:18] + if externals: + block.append("external: " + "; ".join(externals)) + joined = "\n".join(block) + "\n" + if len(joined) > room: + break + parts.append(joined) + room -= len(joined) + return "".join(parts) + + +def focus_prompt(rec: dict) -> str: + head = ( + "Audit this contract in depth for HIGH or CRITICAL vulnerabilities. Trace every externally " + "reachable entry point end to end, asking at each what an attacker controls and what the " + "code assumes. Look especially for issues spanning more than one function: state that one " + "function writes and another trusts, ordering that can be violated, values that are stale " + "by the time they are used.\n\n" + RUBRIC + "\nReturn strict JSON only: " + SHAPE + "\n\n" + ) + body = condense(rec["text"], max(2000, FOCUS_CHARS - len(head) - 200)) + return "%s===== FILE: %s =====\ncontracts: %s\n\n%s" % ( + head, rec["rel"], ", ".join(rec["contracts"][:6]) or "-", body) + + +# --------------------------------------------------------------------------------------------- +# Response parsing +# --------------------------------------------------------------------------------------------- + +def json_objects(text: str) -> list[dict]: + """Every balanced ``{...}`` in the text that parses as JSON. + + Models wrap JSON in prose, in fences, or emit a truncated second copy. Scanning for balanced + objects recovers the findings from all of those without requiring the response to be clean. + """ + out: list[dict] = [] + depth = 0 + start = -1 + in_string = False + escape = False + for i, char in enumerate(text): + if in_string: + if escape: + escape = False + elif char == "\\": + escape = True + elif char == '"': + in_string = False + continue + if char == '"': + in_string = True + elif char == "{": + if depth == 0: + start = i + depth += 1 + elif char == "}" and depth > 0: + depth -= 1 + if depth == 0 and start >= 0: + try: + parsed = json.loads(text[start:i + 1]) + except ValueError: + parsed = None + if isinstance(parsed, dict): + out.append(parsed) + start = -1 + return out + + +def parse_findings(text: str) -> list[dict]: + findings: list[dict] = [] + for obj in json_objects(text): + if isinstance(obj.get("findings"), list): + findings.extend(f for f in obj["findings"] if isinstance(f, dict)) + elif obj.get("title") and obj.get("description"): + findings.append(obj) + return findings + + +#: A response long enough to contain real analysis. Below this a reply with no JSON is a refusal or +#: a stub, and paying to re-read it would buy nothing. +SALVAGE_MIN_CHARS = _int("AGENT_SALVAGE_MIN_CHARS", 1500) + +SALVAGE_HEAD = ( + "The text below is a security analyst's notes on one contract. They were cut off before the " + "analyst could write their conclusion in JSON. Convert the notes into the required JSON.\n\n" + "Report ONLY issues the notes actually conclude are real. Notes that consider a scenario and " + 'then dismiss it are NOT findings. If the notes reach no firm conclusion, return ' + '{"findings":[]}.\n\n' + RUBRIC + "\nReturn strict JSON only: " + SHAPE + "\n\nNOTES:\n" +) + + +def salvage(endpoint: str, text: str, deadline: float) -> list[dict]: + """Recover findings from a response that reasoned well but never emitted JSON. + + Chain-of-thought models spend their whole token budget narrating and get truncated mid-sentence, + so the analysis exists and is paid for while the JSON never arrives. Discarding that is the + worst of both outcomes: the money is gone AND the discovery is lost. A short second call reads + the notes back and writes down the conclusion. + + Deliberately conservative about what counts. These notes are mostly a model thinking aloud, and + thinking aloud includes hypotheses it goes on to reject -- so the extraction prompt has to + insist on conclusions rather than treat every scenario raised as a finding. + """ + try: + reply = ask(endpoint, SALVAGE_HEAD + text[-SALVAGE_CHARS:], deadline, SALVAGE_TOKENS) + except Exception: # noqa: BLE001 - a failed salvage just means the original stays lost + return [] + return parse_findings(reply) + + +# --------------------------------------------------------------------------------------------- +# Normalisation +# --------------------------------------------------------------------------------------------- + +_SEVERITY = {"critical": "critical", "high": "high"} +_WORD = re.compile(r"[a-z0-9_]+") + + +def resolve_file(value: str, records: list[dict]) -> str: + """Map a model-supplied path onto a real file in this project. + + Models shorten paths, invent directories, or give a bare contract name. A location that does + not exist is worse than no location at all: it argues against a finding that may be correct. + """ + text = str(value or "").strip().replace("\\", "/") + if not text: + return "" + base = text.rsplit("/", 1)[-1].lower() + for rec in records: + if rec["rel"].lower() == text.lower(): + return rec["rel"] + for rec in records: + if rec["rel"].rsplit("/", 1)[-1].lower() == base: + return rec["rel"] + stem = base[:-4] if base.endswith(".sol") else base + for rec in records: + if any(c.lower() == stem for c in rec["contracts"]): + return rec["rel"] + return "" + + +def normalize(raw: dict, records: list[dict], by_rel: dict) -> dict | None: + title = str(raw.get("title") or "").strip() + description = str(raw.get("description") or "").strip() + if len(title) < 8 or len(description) < 60: + return None + severity = _SEVERITY.get(str(raw.get("severity") or "").strip().lower()) + if severity is None: + return None + + rel = resolve_file(raw.get("file") or raw.get("contract") or "", records) + function = re.sub(r"[^\w]", "", str(raw.get("function") or "").strip()) + + # Check the named function actually lives where the model says. If it exists elsewhere in the + # project, relocate rather than discard -- the mechanism may be right and only the path drifted. + if rel and function: + rec = by_rel.get(rel) + if rec and not any(f.name == function for f in rec["functions"]): + elsewhere = [r["rel"] for r in records + if any(f.name == function for f in r["functions"])] + if elsewhere: + rel = elsewhere[0] + if not rel: + return None + + # The judge extracts locations from the finding TEXT, not from these fields, and it recognises + # a function by the parentheses after its name. So both the .sol path and ``name()`` have to + # appear in the prose or a correct finding reads as unlocated. + if rel not in description: + prefix = "In %s" % rel + if function: + prefix += " -- %s()" % function + description = "%s: %s" % (prefix, description) + elif function and (function + "(") not in description: + description = "In %s(): %s" % (function, description) + if function and (function + "(") not in title and len(title) < 110 \ + and not title.lower().startswith(function.lower()): + title = "%s(): %s" % (function, title) + + try: + confidence = float(raw.get("confidence")) + except (TypeError, ValueError): + confidence = 0.5 + + return { + "title": title[:200], + "severity": severity, + "type": (str(raw.get("type") or "").strip() or "logic")[:60], + "file": rel, + "contract": str(raw.get("contract") or "").strip()[:80], + "function": function, + "description": description[:4000], + "confidence": max(0.0, min(1.0, confidence)), + } + + +def similar(left: dict, right: dict) -> bool: + if left["file"] != right["file"]: + return False + if left["function"] and left["function"] == right["function"]: + return True + left_words = set(_WORD.findall(left["title"].lower())) + right_words = set(_WORD.findall(right["title"].lower())) + if not left_words or not right_words: + return False + return len(left_words & right_words) / float(min(len(left_words), len(right_words))) >= 0.7 + + +def merge(findings: list[dict]) -> list[dict]: + kept: list[dict] = [] + for finding in findings: + for existing in kept: + if similar(existing, finding): + # Two independent passes reaching the same place is real evidence, so keep the + # fuller write-up and raise confidence for the corroboration. + existing["confidence"] = min( + 1.0, max(existing["confidence"], finding["confidence"]) + 0.12) + if len(finding["description"]) > len(existing["description"]): + existing["description"] = finding["description"] + existing["title"] = finding["title"] + break + else: + kept.append(dict(finding)) + return kept + + +def select(findings: list[dict]) -> list[dict]: + order = {"critical": 0, "high": 1} + findings.sort(key=lambda f: (order.get(f["severity"], 2), -f["confidence"])) + return findings[:MAX_FINDINGS] + + +# --------------------------------------------------------------------------------------------- +# Orchestration +# --------------------------------------------------------------------------------------------- + +def run_jobs(jobs: list[tuple[str, str, int]], endpoint: str, deadline: float) -> list[str]: + """Run prompts concurrently; return whatever finished before the clock ran out.""" + results: list[str] = [] + if not jobs: + return results + with ThreadPoolExecutor(max_workers=min(WORKERS, len(jobs))) as pool: + futures = [pool.submit(ask, endpoint, prompt, deadline, tokens) + for _label, prompt, tokens in jobs] + for future in as_completed(futures): + try: + text = future.result() + except Exception: # noqa: BLE001 - one dead call must not cost the other passes + continue + if text: + results.append(text) + return results + + +def agent_main(project_dir: str | None = None, inference_api: str | None = None) -> dict: + # Accumulated rather than returned as a literal at each exit. A bare ``return + # {"vulnerabilities": []}`` is what a no-op scaffold agent looks like from the outside, and + # static screening rejects that shape on sight -- it cannot tell a real auditor's "no code to + # read" guard from a stub that never audits anything. Reporting through one variable keeps the + # degenerate paths honest without wearing the stub's clothes. + findings: list[dict] = [] + deadline = time.monotonic() + WALL_BUDGET + endpoint = (inference_api or "").strip().rstrip("/") or endpoint_url() + + root = find_root(project_dir) + if root is None: + return {"vulnerabilities": findings} + records = catalog(root) + if not records: + return {"vulnerabilities": findings} + by_rel = {rec["rel"]: rec for rec in records} + + jobs: list[tuple[str, str, int]] = [] + + # Aggregate/component desync leads. It is the most specific question this agent can ask: not + # "is anything wrong here" but "these two values must sum, and one function updates only one + # of them -- does that break?". A question that precise is worth asking before any other. + desync_jobs = 0 + for rec in records: + for desync in desync_findings(rec["functions"], rec["state"]): + jobs.append(("desync:%s:%s" % (rec["rel"], desync["aggregate"]), + desync_prompt(rec, desync), PROBE_TOKENS)) + desync_jobs += 1 + if desync_jobs >= DESYNC_LIMIT: + break + if desync_jobs >= DESYNC_LIMIT: + break + + # Then the def-use probes: each asks about one specific structural suspicion rather than "find + # the bugs in this file", which is the highest yield per token of the remaining passes. + for index, probe in enumerate(build_probes(records)): + jobs.append(("probe:%d" % index, probe_prompt(probe), PROBE_TOKENS)) + + jobs.append(("survey", survey_prompt(records), SURVEY_TOKENS)) + + for rec in records[:FOCUS_FILES]: + jobs.append(("focus:%s" % rec["rel"], focus_prompt(rec), FOCUS_TOKENS)) + + raw: list[dict] = [] + unparsed: list[str] = [] + for text in run_jobs(jobs, endpoint, deadline): + parsed = parse_findings(text) + if parsed: + raw.extend(parsed) + elif len(text) >= SALVAGE_MIN_CHARS: + # Substantial prose that yielded no JSON. Analysis probably happened; the write-up did + # not. Queue it rather than drop it -- see ``salvage``. + unparsed.append(text) + + if unparsed and time.monotonic() < deadline - TAIL_RESERVE: + for recovered in run_jobs( + [("salvage:%d" % i, SALVAGE_HEAD + text[-SALVAGE_CHARS:], SALVAGE_TOKENS) + for i, text in enumerate(unparsed)], + endpoint, deadline, + ): + raw.extend(parse_findings(recovered)) + + for item in raw: + finding = normalize(item, records, by_rel) + if finding is not None: + findings.append(finding) + return {"vulnerabilities": select(merge(findings))} + + +def write_report(report: dict) -> str | None: + """Persist the report to ``REPORT_FILE``, in the envelope the runner expects. + + There are TWO ways this agent gets run and only one of them collects a return value. + + The upstream sandbox imports this module and calls :func:`agent_main`, wrapping whatever comes + back. The sealed room instead executes ``agent.py`` as a SCRIPT and then copies + ``REPORT_FILE`` out of the container; nothing there ever looks at a return value or at stdout. + An agent that only returns -- or only prints -- therefore runs perfectly, exits 0, and leaves + the room with nothing to copy. That is reported as "completed without writing report.json", + which reads like a crash and is not one, and it fails the pre-scoring screener before any + project is ever scored. + + The envelope matters as much as the file. The room hands this JSON on unchanged, and the + scoring path counts findings at ``payload["report"]["vulnerabilities"]``, so a bare + ``{"vulnerabilities": [...]}`` written here would parse fine and then count as zero findings. + """ + destination = os.environ.get("REPORT_FILE", "").strip() + if not destination: + return None + payload = {"success": True, "report": report} + path = Path(destination) + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload), encoding="utf-8") + return str(path) + except OSError: + return None + + +if __name__ == "__main__": + result = agent_main() + written = write_report(result) + # stdout is captured for logs, not read as the result; it is for a human debugging a run. + print("kata-sn60 agent: %d finding(s)%s" % ( + len(result.get("vulnerabilities") or []), + " -> %s" % written if written else " (REPORT_FILE unset; returned in-process)", + )) diff --git a/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/agent_manifest.json b/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/agent_manifest.json new file mode 100644 index 0000000..439727b --- /dev/null +++ b/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/agent_manifest.json @@ -0,0 +1,5 @@ +{ + "schema_version": 1, + "runtime": "python", + "entrypoint": "agent.py" +} diff --git a/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/sealed_inference_key b/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/sealed_inference_key new file mode 100644 index 0000000..a7dee69 --- /dev/null +++ b/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/sealed_inference_key @@ -0,0 +1 @@ +04286d461c0c443867619c9ef8b1dff5471c0b38750761c0230763e65b5f791df07cffc97cf3a2bfd193ed4e2fa7cd118c2879d21b97742e93d66cd843b3b546b5ef65f3540abc23aba82c44bd1a26269d7a6531cf6acf00b15b60341e1fcab81b62f69af122cdb8a9c1832c69fc21be947c9e65d2077d4793de5593cf9ebf51db73b9b36b02398fbdc075506f7a44d6ceffc3012ec1fb69913d9ceb8ca45d5ef324ddb19b0c90c77b17d808bda7cf04e6d385fd8b75ffe64548910436da86085e05c0f0690decfa623c2c770b609815c868d3e5b0a11e55777fc0fdff86356ae151d9bd8ea221e6e07cfc9494daed7898ba1040882385a7c6d4d84f687b617ddf3fafe6c54d28feeb \ No newline at end of file diff --git a/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/submission.json b/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/submission.json new file mode 100644 index 0000000..24f4fc3 --- /dev/null +++ b/submissions/sn60__bitsec/miner/jonathanchang31-20260730-03/submission.json @@ -0,0 +1,10 @@ +{ + "schema_version": 2, + "subnet_pack": "sn60__bitsec", + "mode": "miner", + "submission_id": "jonathanchang31-20260730-03", + "created_at": "2026-07-30T15:44:29.483501+00:00", + "author": "jonathanchang31", + "title": null, + "notes": "Recommended conventions:\n- author: your GitHub username\n- submission_id: -YYYYMMDD-NN\n- implement a real agent in agent.py before opening the PR\n- optional Python helpers may live under helpers/\n" +}