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
12 changes: 12 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -1076,6 +1076,18 @@ jobs:
|| { echo "FAIL: expected exactly one OWN001 (the uncleared timer)"; exit 1; }
echo "$edges" | grep -q "pollB" \
|| { echo "FAIL: the leak must be the second (uncleared) interval"; exit 1; }
- name: Parser hardening — string literals, nested dep brackets, listener options
run: |
python frontend/ownts/ownts.py frontend/ownts/examples/EffectHardening.tsx \
-o "$RUNNER_TEMP/hard.facts.json"
hard=$(python -m ownlang ownir "$RUNNER_TEMP/hard.facts.json" || true)
echo "$hard"
# a string with commas/braces does not truncate the body (the timer is cleared);
# the leak is the options-dropped listener; the object dep fires EFF001 once
[ "$(echo "$hard" | grep -cE '\[(OWN|EFF|DI)[0-9]{3}\]')" -eq 2 ] \
|| { echo "FAIL: expected one OWN001 (listener) + one EFF001 (object dep)"; exit 1; }
echo "$hard" | grep -q "scroll" \
|| { echo "FAIL: the leak must be the options-dropped scroll listener"; exit 1; }
- name: The clean fixture (cleanups + useMemo'd dep) is silent
run: |
python frontend/ownts/ownts.py frontend/ownts/examples/DashboardClean.tsx \
Expand Down
31 changes: 31 additions & 0 deletions frontend/ownts/examples/EffectHardening.tsx
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
// Parser-hardening cases (CodeRabbit): string literals must not truncate bodies or
// split deps, deps brackets nest, and a listener is released only by a matching
// target+handler+options removeEventListener. Run:
// python frontend/ownts/ownts.py frontend/ownts/examples/EffectHardening.tsx --check
// Expect exactly: OWN001 (the options-dropped listener) + EFF001 (the object dep).
import { useEffect } from "react";

export function Hardening({ id }: { id: string }) {
// A string with commas and braces must not truncate the body or split the deps;
// the timer IS cleared, so this effect leaks nothing.
useEffect(() => {
const t = setInterval(() => fetch("/a,b,{c}"), 1000);
return () => clearInterval(t);
}, [id]);

// Listener added WITH capture, but cleanup drops the option -> different listener,
// still leaks (one OWN001).
useEffect(() => {
window.addEventListener("scroll", onScroll, true);
return () => window.removeEventListener("scroll", onScroll);
}, []);

// Nested-bracket dep parses (`items[0]` stays one entry); the object literal dep
// is unstable + IO -> exactly one EFF001 (items[0] is not identifier-like: silent).
const filters = { id };
useEffect(() => {
fetch("/x");
}, [filters, items[0]]);

return <div>hardening</div>;
}
238 changes: 190 additions & 48 deletions frontend/ownts/ownts.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,45 @@ def _lhs_token(setup: str, pos: int) -> str | None:
return m.group(1) if m else None


def _capture_flag(opts: str) -> str:
"""The capture flag an addEventListener/removeEventListener options arg implies —
the only field that identifies a listener for removal. `'true'`/`'false'`, or
`'unknown'` for a non-literal we cannot read. An omitted arg, or an options
object without a `capture` key, defaults to capture **false** (the DOM default);
`passive`/`once`/`signal` do not affect removal identity."""
o = opts.strip()
if o in ("", "false"):
return "false"
if o == "true":
return "true"
if o.startswith("{"):
m = re.search(r"\bcapture\s*:\s*(true|false)\b", o)
return m.group(1) if m else "false"
return "unknown" # a variable/call — compare verbatim (equal only to itself)


_LISTENER = re.compile(
r"\.\s*(?:add|remove)EventListener\s*\(\s*([^,]+?)\s*,\s*([A-Za-z_$][\w$.]*)\s*"
r"(?:,\s*([^)]+?))?\s*\)")


def _listener_call(s: str) -> tuple[str, str, str] | None:
"""Parse a `.addEventListener`/`.removeEventListener` head into the
(event, handler, capture) triple that identifies the listener for removal, or
None when it doesn't parse. `s` starts at the `.`."""
m = _LISTENER.match(s.lstrip())
if not m:
return None
return (m.group(1).strip(), m.group(2), _capture_flag(m.group(3) or ""))


def _receiver(text: str, end: int) -> str:
"""The receiver expression ending at `end` (just before `.addEventListener`),
allowing a member/index chain like `this.ref` or `nodes[i]`."""
m = re.search(r"([A-Za-z_$][\w$.\[\]]*)$", text[:end])
return m.group(1) if m else ""


def _is_released(acq: Acquire, setup: str, pos: int, cleanup: str) -> bool:
"""Whether THIS acquire (at `pos` in `setup`) is released by the effect's cleanup
— matched to its own handle, so two `setInterval`s with one `clearInterval` leave
Expand All @@ -88,12 +127,26 @@ def _is_released(acq: Acquire, setup: str, pos: int, cleanup: str) -> bool:
tok = _lhs_token(setup, pos)
return bool(tok and re.search(
rf"\b{re.escape(tok)}\s*\.\s*(?:unsubscribe|remove)\s*\(", cleanup))
if acq.resource == "subscription": # addEventListener(event, handler)
hm = re.search(r"addEventListener\s*\(\s*[^,]+,\s*([A-Za-z_$][\w$.]*)",
setup[pos:])
handler = hm.group(1) if hm else None
return bool(handler and re.search(
rf"removeEventListener\s*\(\s*[^,]+,\s*{re.escape(handler)}\b", cleanup))
if acq.resource == "subscription": # target.addEventListener(event, handler[, opts])
# The full listener key is (receiver, event, handler, capture). A listener is
# released ONLY by a removeEventListener whose whole key matches — a different
# event name, target, or capture flag is a different listener that still
# leaks. Fail closed: if the acquire's own target can't be identified, or no
# cleanup call matches every field, treat it as a leak.
a = _listener_call(setup[pos:])
if a is None:
return False
a_recv = _receiver(setup, pos)
if not a_recv:
return False
for rm in re.finditer(r"([A-Za-z_$][\w$.\[\]]*)\s*\.\s*removeEventListener\s*\(",
cleanup):
if rm.group(1) != a_recv:
continue
b = _listener_call(cleanup[rm.start() + len(rm.group(1)):])
if b == a:
return True
return False
return False

# A React component is a function whose name is Capitalized (the JSX convention).
Expand Down Expand Up @@ -206,6 +259,68 @@ def _match_block(text: str, open_idx: int) -> int:
return len(text)


def _mask_strings(text: str) -> str:
"""Blank the CONTENT of string/template literals — keeping the delimiters, the
length, and every newline — so the structural scanners (brace/paren/bracket
depth, top-level commas, the deps array) are never fooled by punctuation inside
a literal: `fetch("/a,b")`, `const s = "{"`, a `;` inside a string. Comments are
already gone (`_strip_comments`). Positions are preserved, so a match found on
the masked copy slices identically out of the original text."""
out = list(text)
i, n = 0, len(text)
while i < n:
c = text[i]
if c in "\"'`":
i += 1
while i < n and text[i] != c:
if text[i] == "\\" and i + 1 < n:
for k in (i, i + 1):
if text[k] != "\n":
out[k] = " "
i += 2
continue
if text[i] != "\n":
out[i] = " "
i += 1
i += 1 # past the closing delimiter (kept)
else:
i += 1
return "".join(out)


def _match_pair(text: str, i: int, open_c: str, close_c: str) -> int:
"""Index just past the `close_c` matching the `open_c` at/after `i`. Assumes
string contents are already masked (so no quote-skipping is needed here)."""
i = text.index(open_c, i)
depth = 0
while i < len(text):
if text[i] == open_c:
depth += 1
elif text[i] == close_c:
depth -= 1
if depth == 0:
return i + 1
i += 1
return len(text)


def _split_top_commas(s: str) -> list[str]:
"""Split on commas at bracket/paren/brace depth 0, so a dependency like
`items[i]` or `f(a, b)` stays one entry instead of being torn at its inner comma."""
parts: list[str] = []
depth, start = 0, 0
for j, c in enumerate(s):
if c in "([{":
depth += 1
elif c in ")]}":
depth -= 1
elif c == "," and depth == 0:
parts.append(s[start:j])
start = j + 1
parts.append(s[start:])
return [p.strip() for p in parts if p.strip()]


def _component_at(text: str, idx: int) -> str:
"""Name of the React component enclosing position `idx` — the nearest preceding
Capitalized function declaration. Falls back to a synthetic name."""
Expand Down Expand Up @@ -233,61 +348,85 @@ def _expr_end(text: str, i: int) -> int:
return i


def _effect_callback(text: str, after_open: int) -> tuple[str, int, list[str] | None, int]:
"""Parse `useEffect(<cb>, [deps])` from just past the `(`. Returns
def _effect_callback(masked: str, after_open: int) -> tuple[str, int, list[str] | None, int]:
"""Parse `useEffect(<cb>, [deps])` from just past the `(`, over the STRING-MASKED
source (so literals never truncate a body or split a dep). Returns
(body, body_start, deps, end). Handles BOTH a block `() => { ... }` and an
expression `() => fetch(url)` callback — calling `_match_block` blindly on the
latter would jump to an unrelated `{` or run off the end. `deps` is None when no
dependency array is present; `body` includes the braces for a block callback."""
arrow = text.find("=>", after_open)
dependency array is present; the array is matched by BALANCED brackets so a dep
like `items[i]` survives. `body` includes the braces for a block callback."""
arrow = masked.find("=>", after_open)
i = (arrow + 2) if arrow != -1 else after_open
while i < len(text) and text[i] in " \t\r\n":
while i < len(masked) and masked[i] in " \t\r\n":
i += 1
if i < len(text) and text[i] == "{":
end = _match_block(text, i)
if i < len(masked) and masked[i] == "{":
end = _match_block(masked, i)
else:
end = _expr_end(text, i)
body = text[i:end]
deps_m = re.search(r"\s*,\s*\[([^\]]*)\]", text[end:end + 200])
deps = ([d.strip() for d in deps_m.group(1).split(",") if d.strip()]
if deps_m else None)
end = _expr_end(masked, i)
body = masked[i:end]
# the dependency array is the balanced `[ ... ]` after an optional `, `
deps: list[str] | None = None
j = end
while j < len(masked) and masked[j] in " \t\r\n":
j += 1
if j < len(masked) and masked[j] == ",":
j += 1
while j < len(masked) and masked[j] in " \t\r\n":
j += 1
if j < len(masked) and masked[j] == "[":
close = _match_pair(masked, j, "[", "]")
deps = _split_top_commas(masked[j + 1:close - 1])
return body, i, deps, end


def _split_cleanup(body: str) -> tuple[str, str]:
"""Split an effect block body into (setup, cleanup). Cleanup is the block of the
effect's OWN top-level `return () => { ... }` — a `return ... =>` nested inside a
callback (brace-depth > 1) is NOT the effect cleanup and must not suppress a
leak. For an expression-bodied effect there is no cleanup."""
for m in re.finditer(r"return\s*(?:\(\s*\)|\w+)\s*=>", body):
prefix = body[:m.start()]
def _cleanup_span(mbody: str) -> tuple[int, int] | None:
"""The (start, end) span of the effect's OWN top-level cleanup within `mbody`
(the masked body), or None. A `return ... =>` nested inside a callback
(brace-depth > 1) is NOT the effect cleanup and must not suppress a leak. The
span is returned (not slices) so the caller can cut it out of BOTH the masked
body (for acquire scanning) and the original body (for event/string comparison)."""
for m in re.finditer(r"return\s*(?:\(\s*\)|\w+)\s*=>", mbody):
prefix = mbody[:m.start()]
if prefix.count("{") - prefix.count("}") != 1: # 1 == the effect body's own brace
continue
brace = body.find("{", m.end())
brace = mbody.find("{", m.end())
if brace == -1:
# `return () => clearInterval(id)` — single-expression cleanup, no block.
nl = body.find("\n", m.end())
tail = body[m.end(): nl if nl != -1 else len(body)]
return body[: m.start()], tail
end = _match_block(body, brace)
return body[: m.start()] + body[end:], body[brace:end]
return body, ""
nl = mbody.find("\n", m.end())
return (m.start(), nl if nl != -1 else len(mbody))
return (m.start(), _match_block(mbody, brace))
Comment on lines +393 to +398

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Expression-bodied cleanup with an options object is mis-spanned, causing false-positive leaks.

brace = mbody.find("{", m.end()) searches arbitrarily far for the cleanup block's {. For a single-expression cleanup whose removeEventListener passes an options object — exactly the case this PR targets — the { of the options literal is found instead:

return () => el.removeEventListener("scroll", onScroll, {capture: true})

_match_block then ends the span at the options }, truncating the trailing ). The resulting cleanup_o no longer parses via _LISTENER (which requires a closing )), so _listener_call returns None, _is_released returns False, and a correctly-released listener is reported as a leak. Block-bodied cleanups are unaffected; only expression bodies containing braces break.

Detect a block only when the first non-whitespace char after => is {:

🐛 Proposed fix
-        brace = mbody.find("{", m.end())
-        if brace == -1:
-            # `return () => clearInterval(id)` — single-expression cleanup, no block.
-            nl = mbody.find("\n", m.end())
-            return (m.start(), nl if nl != -1 else len(mbody))
-        return (m.start(), _match_block(mbody, brace))
+        rest = mbody[m.end():]
+        stripped = rest.lstrip()
+        if stripped.startswith("{"):  # block-bodied cleanup `=> { ... }`
+            brace = m.end() + (len(rest) - len(stripped))
+            return (m.start(), _match_block(mbody, brace))
+        # single-expression cleanup, e.g. `return () => clearInterval(id)` or
+        # `return () => el.removeEventListener("x", h, {capture: true})`
+        nl = mbody.find("\n", m.end())
+        return (m.start(), nl if nl != -1 else len(mbody))
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
brace = mbody.find("{", m.end())
if brace == -1:
# `return () => clearInterval(id)` — single-expression cleanup, no block.
nl = body.find("\n", m.end())
tail = body[m.end(): nl if nl != -1 else len(body)]
return body[: m.start()], tail
end = _match_block(body, brace)
return body[: m.start()] + body[end:], body[brace:end]
return body, ""
nl = mbody.find("\n", m.end())
return (m.start(), nl if nl != -1 else len(mbody))
return (m.start(), _match_block(mbody, brace))
rest = mbody[m.end():]
stripped = rest.lstrip()
if stripped.startswith("{"): # block-bodied cleanup `=> { ... }`
brace = m.end() + (len(rest) - len(stripped))
return (m.start(), _match_block(mbody, brace))
# single-expression cleanup, e.g. `return () => clearInterval(id)` or
# `return () => el.removeEventListener("x", h, {capture: true})`
nl = mbody.find("\n", m.end())
return (m.start(), nl if nl != -1 else len(mbody))
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@frontend/ownts/ownts.py` around lines 393 - 398, The cleanup-span detection
in the expression-bodied branch is too permissive and mistakenly treats an
options object inside a call as the cleanup block. In the span logic around
mbody.find("{", m.end()) and _match_block, only classify a cleanup as
block-bodied when the first non-whitespace character after => is {; otherwise
keep the expression-body range ending at the line break. This preserves calls
like removeEventListener(..., {capture: true}) so _listener_call and
_is_released continue to parse them correctly.

return None


def extract(path: str) -> list[Component]:
text = _strip_comments(open(path, encoding="utf-8").read())
masked = _mask_strings(text) # scan structure on this; show snippets from `text`
comps: dict[str, Component] = {}
for eff in _USE_EFFECT.finditer(text):
body, body_start, _deps, _end = _effect_callback(text, eff.end())
setup, cleanup = _split_cleanup(body)
cname = _component_at(text, eff.start())
for eff in _USE_EFFECT.finditer(masked):
body_m, body_start, _deps, end = _effect_callback(masked, eff.end())
body_o = text[body_start:end] # original (unmasked) body — same positions
span = _cleanup_span(body_m)
if span:
cs, ce = span
setup_m = body_m[:cs] + body_m[ce:]
setup_o = body_o[:cs] + body_o[ce:]
cleanup_o = body_o[cs:ce]
else:
setup_m, setup_o, cleanup_o = body_m, body_o, ""
cname = _component_at(masked, eff.start())
comp = comps.setdefault(cname, Component(cname, path))
for acq in ACQUIRES:
for hit in acq.pattern.finditer(setup):
line = text.count("\n", 0, body_start + hit.start()) + 1
released = _is_released(acq, setup, hit.start(), cleanup)
# the acquire expression, trimmed to the call head for a readable tag
snippet = setup[hit.start():].splitlines()[0].strip().rstrip("{").strip()
# find the acquire on the MASKED setup (a keyword inside a string is gone);
# decide release on the ORIGINAL setup/cleanup (so event-name strings and
# handles are intact for the full-key comparison).
for hit in acq.pattern.finditer(setup_m):
abs_pos = body_start + hit.start()
line = masked.count("\n", 0, abs_pos) + 1
released = _is_released(acq, setup_o, hit.start(), cleanup_o)
# the acquire expression for the tag — from the ORIGINAL text, so the
# message shows the real call (string args intact), trimmed to one line.
snippet = text[abs_pos:].splitlines()[0].strip().rstrip("{").strip()
comp.resources.append(
Resource(snippet or acq.name, line, released, acq.resource, acq.eff))
return [c for c in comps.values() if c.resources]
Expand Down Expand Up @@ -361,8 +500,10 @@ def _classify_rhs(rhs: str) -> tuple[str, list[str]]:


def _render_bindings(text: str) -> dict[str, list[dict]]:
"""The render-scope binding table per component: only bindings declared DIRECTLY
in the component body (brace-depth 1). A `const filters = {...}` inside a
"""The render-scope binding table per component, scanned over STRING-MASKED
source so a brace/quote inside a literal cannot shift the depth. Only bindings
declared DIRECTLY in the component body (brace-depth 1) count. A `const filters
= {...}` inside a
`useEffect` callback, an event handler, or any nested block is NOT render scope —
it must not shadow the real outer dependency of the same name and mint a false
EFF001. Excluding a render-level `if`/`try` binding only costs a missed finding
Expand Down Expand Up @@ -393,16 +534,17 @@ def extract_effects(path: str) -> list[dict]:
whether its body does network IO, and the render-scope binding table of the
component it lives in. The core's effects analysis turns these into a verdict."""
text = _strip_comments(open(path, encoding="utf-8").read())
binds_by_comp = _render_bindings(text)
masked = _mask_strings(text)
binds_by_comp = _render_bindings(masked)
effects: list[dict] = []
for eff in _USE_EFFECT.finditer(text):
body, _start, deps, _end = _effect_callback(text, eff.end())
for eff in _USE_EFFECT.finditer(masked):
body, _start, deps, _end = _effect_callback(masked, eff.end())
if deps is None: # no dep array -> not an EFF001 candidate (by-design re-run cadence)
continue
cname = _component_at(text, eff.start())
cname = _component_at(masked, eff.start())
effects.append({
"component": cname, "file": path,
"line": text.count("\n", 0, eff.start()) + 1,
"line": masked.count("\n", 0, eff.start()) + 1,
"io": bool(_IO.search(body)),
"deps": deps,
"bindings": binds_by_comp.get(cname, []),
Expand Down
Loading
Loading