fix(scan,taint): TypeScript's limit is owned by TypeScript; a source pattern is a token

Two defects found by the cause-B rescan of the independent population (exploratory,
a retune after protocol-v1):

- `typescript_property_support_unvalidated` still reached the whole repository.
  YesWiki is a PHP application that ships a few `.ts` files, and all 1,128 of its PHP
  questions were unresolved. The limit is now owned by the TypeScript partition. A
  TypeScript project (Mockoon, Signal K) is unchanged.
- `isSourceCall`, and the method-level "reads a source" heuristics, matched source
  patterns as substrings. Flask's `request.get` therefore claimed Django's
  `request.get_host()`. wger's fix passes that host to
  `url_has_allowed_host_and_scheme`, and both fixed functions were reported as open
  redirects from it. Patterns are now matched as bounded tokens, like sinks and
  sanitizers. wger's vulnerable-pin detections rest on `request.GET`, which still
  matches.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

src/openultrasast/cpg/queries/taint.sc

@@ -174,7 +174,10 @@
       if (accessOperators.contains(c.name)) c.code
       else if (c.name.startsWith("<operator")) ""
       else c.code.takeWhile(_ != '(')
-    named.nonEmpty && sourcePatterns.exists(p => named.contains(p))
+    // Bounded, like every other token here: a substring test let Flask's `request.get` claim Django's
+    // `request.get_host()` -- the host handed to the redirect VALIDATOR in wger's fix -- and both fixed
+    // functions were reported as open redirects.
+    named.nonEmpty && sourcePatterns.exists(p => mentionsToken(named, List(p)))
   }
 
   def frameworkSources = cpg.call.filter(isSourceCall)
@@ -1003,7 +1006,7 @@
   def sourceMethods(fileName: String): Set[String] =
     sourceMethodsMemo.getOrElseUpdate(
       fileName,
-      methodsIn(fileName).filter(m => m.ast.isCall.exists(c => sourcePatterns.exists(p => c.code.contains(p)))).map(_.name).toSet
+      methodsIn(fileName).filter(m => m.ast.isCall.exists(c => mentionsToken(c.code, sourcePatterns))).map(_.name).toSet
     )
 
   def fedFields(fileName: String): Map[String, String] =
@@ -1023,7 +1026,7 @@
                     case id: io.shiftleft.codepropertygraph.generated.nodes.Identifier => params.contains(id.name)
                     case _                                                             => false
                   }
-                  val fromSource = sourcePatterns.exists(p => rhs.code.contains(p)) || (readers.nonEmpty && rhs.ast.isCall.name.l.exists(readers.contains))
+                  val fromSource = mentionsToken(rhs.code, sourcePatterns) || (readers.nonEmpty && rhs.ast.isCall.name.l.exists(readers.contains))
                   if (fromSource) Some(target.code.trim -> "source")
                   else if (fromParameter) Some(target.code.trim -> "parameter")
                   else None
@@ -1041,7 +1044,7 @@
     callbackFedMemo.getOrElseUpdate(
       callback,
       strongest(cpg.method.nameExact(callback).l.map { m =>
-        if (m.ast.isCall.exists(c => sourcePatterns.exists(p => c.code.contains(p)))) "source"
+        if (m.ast.isCall.exists(c => mentionsToken(c.code, sourcePatterns))) "source"
         else {
           val fed = fedFields(m.filename)
           if (fed.isEmpty) "" else strongest(m.ast.isCall.nameExact(FIELD_ACCESS).code.l.map(code => prefixKind(code.trim, fed)))
@@ -1073,7 +1076,7 @@
     // no-dataflow query take minutes. `cpg.method.filter(...)` over every method per request was the
     // other half of that, replaced by a direct lookup of the reachable names.
     def methodHasSource(m: io.shiftleft.codepropertygraph.generated.nodes.Method): Boolean =
-      methodSourceMemo.getOrElseUpdate(m.fullName, m.ast.isCall.exists(c => sourcePatterns.exists(p => c.code.contains(p))))
+      methodSourceMemo.getOrElseUpdate(m.fullName, m.ast.isCall.exists(c => mentionsToken(c.code, sourcePatterns)))
     val sinkList    = sinkCalls.l
     val sourceLocal = labeledMethods.exists(methodHasSource)
     val sourceNear  = depth > 0 && cpg.method.fullNameExact(reachableMethods.toSeq: _*).exists(methodHasSource)

src/openultrasast/cpg/queries/vocabulary.sc

@@ -54,7 +54,7 @@
     else
       sourceMemo.getOrElseUpdate(
         c.method.fullName,
-        c.method.ast.isCall.exists(x => sourcePatterns.exists(p => x.code.contains(p)))
+        c.method.ast.isCall.exists(x => sourcePatterns.exists(p => boundedPattern(p).matcher(x.code).find()))
       )
 
   def isModelled(c: io.shiftleft.codepropertygraph.generated.nodes.Call): Boolean = {

src/openultrasast/model/partitions.py

@@ -279,6 +279,9 @@ def build_partitions(
             if language == "typescript":
                 record_limits += ("typescript_property_support_unvalidated",)
                 boundaries.add("typescript_property_support_unvalidated")
+                # About TypeScript answers. YesWiki is a PHP application that ships a few `.ts` files, and this
+                # boundary made all 1,128 of its PHP questions unresolved.
+                owned_languages.setdefault("typescript_property_support_unvalidated", set()).add("typescript")
             if language in {"c", "cpp"}:
                 record_limits += ("c_bounds_arithmetic_unmodeled",)
             records.append(

tests/test_cpg_queries_compile.py

@@ -1228,3 +1228,56 @@ def test_an_escaped_but_unquoted_field_is_object_state_that_carries_input(tmp_pa
     }
     assert any("$this->order" in str(r.get("source")) for r in unsanitized["members"]), payload["members"]
     assert not unsanitized["named"], f"a field built from a quoted escape was reported: {unsanitized['named']}"
+
+
+@pytest.mark.skipif(_joern() is None, reason="joern is not installed on this machine")
+def test_a_source_pattern_is_a_token_not_a_prefix(tmp_path: Path) -> None:
+    """Flask's `request.get` matched Django's `request.get_host()` as a substring. wger's fix hands the host to
+    `url_has_allowed_host_and_scheme`, and both fixed functions were reported as open redirects from it. The
+    real parameter, `request.GET.get("next")`, is still a source."""
+    from openultrasast.cpg.backend import extract_payload
+    from openultrasast.model.specs import taint_specs
+
+    frontend = __import__("shutil").which("pysrc2cpg")
+    if frontend is None:
+        pytest.skip("pysrc2cpg is not installed")
+    spec = taint_specs(language="python")["untrusted_destination"]
+    assert "request.get" in spec.sources, "the shipped vocabulary no longer holds the pattern this control is about"
+    src = tmp_path / "src"
+    src.mkdir()
+    (src / "views.py").write_text(
+        "from django.shortcuts import redirect\n\n"
+        "def host(request):\n"
+        "    return redirect(request.get_host())\n\n"
+        "def parameter(request):\n"
+        '    return redirect(request.GET.get("next"))\n'
+    )
+    cpg = tmp_path / "cpg.bin"
+    built = subprocess.run([frontend, str(src), "-o", str(cpg)], capture_output=True, text=True, timeout=600, check=False)
+    if not cpg.is_file():
+        pytest.skip(f"could not build a sample cpg: {(built.stderr or '')[-200:]}")
+    common = {"sources": ",".join(spec.sources), "sinks": ",".join(spec.sinks), "file": "views.py"}
+    requests = tmp_path / "requests.json"
+    requests.write_text(json.dumps({name: {**common, "function": name} for name in ("host", "parameter")}))
+    done = subprocess.run(
+        [
+            _joern() or "joern",
+            "--script",
+            str((QUERIES / "taint.sc").resolve()),
+            "--param",
+            f"cpgFile={cpg}",
+            "--param",
+            f"requestsFile={requests}",
+        ],
+        capture_output=True,
+        text=True,
+        timeout=900,
+        check=False,
+        cwd=str(tmp_path),
+    )
+    assert done.returncode == 0, (done.stderr or done.stdout or "")[-500:]
+    payload = extract_payload(done.stdout or "")
+    assert isinstance(payload, dict)
+    flows = {name: [r for r in payload[name] if isinstance(r, dict) and r.get("sourceKind")] for name in ("host", "parameter")}
+    assert not flows["host"], f"request.get_host() was read as request.get: {flows['host']}"
+    assert flows["parameter"], "the request parameter is no longer a source"

tests/test_model_scope.py

@@ -453,3 +453,18 @@ def test_a_language_no_frontend_reads_demotes_only_its_own_questions(tmp_path):
     assert gap and gap[0].get("census_language") == "go", result.degradations
     assert "frontend_unsupported" in result.scope.unresolved_boundaries
     assert result.question_outcomes[0].status == "completed", result.question_outcomes
+
+
+def test_unvalidated_typescript_demotes_only_typescript_questions(tmp_path):
+    """YesWiki is PHP with a few `.ts` files; the TypeScript limit made every one of its PHP questions unresolved."""
+    root = source(tmp_path, "app.py", "widget.ts")
+    result = scan_repository(
+        root,
+        [region("app.py"), region("widget.ts", "typescript")],
+        backend=Backend({"app.py": vector(), "widget.ts": vector()}),
+        population_complete=True,
+    )
+    gap = [d for d in result.degradations if d.get("reason") == "typescript_property_support_unvalidated"]
+    assert gap and gap[0].get("census_language") == "typescript", result.degradations
+    status = {o.identity.path: o.status for o in result.question_outcomes}
+    assert status == {"app.py": "completed", "widget.ts": "unresolved"}, status