fix(gate-48): a net-zero MOVE is not a removal, and 'no co-change in the diff' is not 'no co-change needed'

hydra-gates/scripts/lib/check_csrf_callers.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: EUPL-1.2
+"""Gate-48 companion — which frontend call sites send NO CSRF token?
+
+WHY THIS EXISTS
+---------------
+Gate-48 asks one question of a diff that removes ``@NoCSRFRequired``: *did the
+same diff add a CSRF signal under* ``src/``? For a PR whose callers have
+**always** sent a token there is no such signal to add, and the gate cannot be
+satisfied without a waiver.
+
+Measured on ConductionNL/larpingapp#298, which closes a live CSRF-forgery hole:
+``SettingsController::create()`` and ``reimport()`` carried
+
+    * @NoCSRFRequired removed to close the CSRF-forgery surface (closes #206).
+
+at docblock-tag position, where Nextcloud's ``ControllerMethodReflector`` reads
+it as the annotation being PRESENT — so the sentence announcing the removal was
+what kept CSRF disabled. Removing it is the fix. All three frontend callers
+already sent ``requesttoken``, and the shared ``CnAdminSettingsShell`` uses
+``@nextcloud/axios``, which injects it. The co-change gate-48 wanted did not
+exist to be made, and the cheapest way to go green would have been a cosmetic
+edit containing the word ``requesttoken`` — the prose-satisfaction the gate
+programme exists to stop.
+
+THE QUESTION THIS ASKS INSTEAD
+------------------------------
+Not "did the diff change a caller?" but "**is any mutating caller unprotected
+right now?**". That is sound in the conservative direction:
+
+* if EVERY mutating call site already carries a CSRF-bearing mechanism, then
+  whichever one reaches the endpoint whose annotation was removed is protected,
+  and enforcing CSRF cannot break it;
+* if ANY mutating call site lacks one, we cannot tell that it is not the
+  caller of that endpoint, so the removal still blocks.
+
+opencatalogi#79 — the defect gate-48 was built for, a delete-modal ``fetch()``
+with no CSRF header — is still caught: that call site is reported here, so the
+gate still fails. A fix that stopped catching it would be a gate switched off.
+
+WHAT COUNTS AS PROTECTED
+------------------------
+Within the call expression itself: ``requesttoken`` / ``OCS-APIRequest`` (both
+case-insensitive, HTTP header names are), or ``getRequestToken``. Or the call
+goes through ``@nextcloud/axios``, imported in that file — that client attaches
+the current token itself, which is the canonical Nextcloud mechanism.
+
+Usage::
+
+    check_csrf_callers.py <app-dir>
+
+Prints one ``path:line — reason`` per UNPROTECTED mutating call site.
+Exits 0 always; the OUTPUT is the answer (#209).
+"""
+from __future__ import annotations
+
+import os
+import re
+import sys
+
+SRC_SUFFIXES = ('.vue', '.js', '.ts', '.mjs', '.cjs')
+SKIP_DIRS = {'node_modules', 'dist', 'build', 'vendor', '.git', 'coverage'}
+
+# `method: 'POST'` / `method: "put"` / `method:\n 'PATCH'` inside a fetch init.
+MUTATING_METHOD = re.compile(
+    r"""method\s*:\s*['"`](?P<verb>POST|PUT|PATCH|DELETE)['"`]""",
+    re.IGNORECASE,
+)
+# axios.post( / axios.put( / this.$axios.delete( ...
+AXIOS_MUTATING = re.compile(
+    r"""\baxios\s*\.\s*(?P<verb>post|put|patch|delete)\s*\(""",
+    re.IGNORECASE,
+)
+FETCH_CALL = re.compile(r"""\bfetch\s*\(""")
+# An import of the Nextcloud axios wrapper, under any local alias.
+NEXTCLOUD_AXIOS_IMPORT = re.compile(
+    r"""from\s+['"]@nextcloud/axios['"]|require\(\s*['"]@nextcloud/axios['"]\s*\)"""
+)
+CSRF_SIGNAL = re.compile(
+    r"""requesttoken|OCS-APIREQUEST|getRequestToken""",
+    re.IGNORECASE,
+)
+
+
+def _call_text(text: str, open_paren: int) -> str:
+    """Text of the call expression starting at the `(` index, paren-balanced.
+
+    Falls back to the rest of the file when the parentheses never balance, so a
+    malformed file reports the call as UNPROTECTED rather than being skipped —
+    an unparseable caller is not evidence of a token.
+    """
+    depth = 0
+    for i in range(open_paren, len(text)):
+        ch = text[i]
+        if ch == '(':
+            depth += 1
+        elif ch == ')':
+            depth -= 1
+            if depth == 0:
+                return text[open_paren:i + 1]
+    return text[open_paren:]
+
+
+def unprotected_call_sites(app_dir: str) -> list[str]:
+    """Mutating frontend call sites carrying no CSRF-bearing mechanism."""
+    findings: list[str] = []
+    src_root = os.path.join(app_dir, 'src')
+    if not os.path.isdir(src_root):
+        return findings
+
+    for root, dirs, files in os.walk(src_root):
+        dirs[:] = [d for d in dirs if d not in SKIP_DIRS]
+        for name in sorted(files):
+            if not name.endswith(SRC_SUFFIXES):
+                continue
+            path = os.path.join(root, name)
+            try:
+                with open(path, encoding='utf-8', errors='replace') as handle:
+                    text = handle.read()
+            except OSError:
+                continue
+            rel = os.path.relpath(path, app_dir)
+            uses_nc_axios = bool(NEXTCLOUD_AXIOS_IMPORT.search(text))
+
+            # 1. axios.<verb>(...) — protected iff the file imports @nextcloud/axios.
+            for m in AXIOS_MUTATING.finditer(text):
+                if uses_nc_axios:
+                    continue
+                call = _call_text(text, m.end() - 1)
+                if CSRF_SIGNAL.search(call):
+                    continue
+                line = text.count('\n', 0, m.start()) + 1
+                findings.append(
+                    f"{rel}:{line} — axios.{m.group('verb').lower()}() with no CSRF "
+                    f"signal and no @nextcloud/axios import"
+                )
+
+            # 2. fetch(...) — mutating iff its init object names a mutating verb.
+            for m in FETCH_CALL.finditer(text):
+                call = _call_text(text, m.end() - 1)
+                verb = MUTATING_METHOD.search(call)
+                if verb is None:
+                    continue
+                if CSRF_SIGNAL.search(call):
+                    continue
+                line = text.count('\n', 0, m.start()) + 1
+                findings.append(
+                    f"{rel}:{line} — fetch() {verb.group('verb').upper()} with no "
+                    f"CSRF signal"
+                )
+    return findings
+
+
+def main(argv: list[str]) -> int:
+    if len(argv) != 2:
+        print("usage: check_csrf_callers.py <app-dir>", file=sys.stderr)
+        return 2
+    for finding in unprotected_call_sites(argv[1]):
+        print(finding)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))

hydra-gates/scripts/lib/check_csrf_removal.py

@@ -67,15 +67,60 @@
 
 # A diff header line is `---` / `---` shaped; it is not a removed line of code.
 DIFF_HEADER = re.compile(r'^---(\s|$)')
+# `+++ b/lib/Controller/X.php` — starts a new file's hunks. `+++` must be
+# tested before the `+` addition branch, exactly as `---` is before `-`.
+DIFF_FILE_HEADER = re.compile(r'^\+\+\+\s+(?:b/)?(?P<path>\S+)')
 
 
 def removals(diff: str) -> list[str]:
-    out = []
+    """Removed lines that genuinely DROPPED CSRF protection.
+
+    A REMOVAL PAIRED WITH AN IDENTICAL ADDITION IS A MOVE, NOT A REMOVAL.
+
+    Relocating a docblock line inside a file emits a `-` and a `+` carrying the
+    same bytes. Scanning only `^-` reads that as deleting the annotation, and
+    the gate reports a security regression for a diff in which nothing about
+    CSRF changed. Measured on larpingapp: #297 moved one comment line while
+    reordering `SettingsController`'s docblocks —
+
+        -     * @NoCSRFRequired removed to close the CSRF-forgery surface (closes #206).
+        +     * instance-wide configuration write needs. `@NoCSRFRequired` was removed
+        +     * @NoCSRFRequired removed to close the CSRF-forgery surface (closes #206).
+
+    — and gate-48 kept `Hydra Gates` red on that repo's `development` branch
+    from then on, over a commit that changed no auth posture at all.
+
+    Cancellation is per FILE and by MULTISET. Per file because a line deleted
+    from one controller and added to another is a real change of posture for
+    the first one; by multiset because a diff that removes a tag twice and
+    restores it once has removed it once.
+    """
+    # {path: [raw content of each added line]} and the removals in file order.
+    added: dict[str | None, list[str]] = {}
+    found: list[tuple[str | None, str]] = []
+    path: str | None = None
+
     for line in diff.splitlines():
+        header = DIFF_FILE_HEADER.match(line)
+        if header:
+            path = header.group('path')
+            continue
+        if line.startswith('+'):
+            added.setdefault(path, []).append(line[1:])
+            continue
         if not line.startswith('-') or DIFF_HEADER.match(line):
             continue
         if ATTRIBUTE_REMOVED.match(line) or DOCBLOCK_TAG_REMOVED.match(line):
-            out.append(line)
+            found.append((path, line))
+
+    out: list[str] = []
+    for file_path, line in found:
+        pool = added.get(file_path)
+        if pool is not None and line[1:] in pool:
+            # Consume the pairing so a second identical removal still reports.
+            pool.remove(line[1:])
+            continue
+        out.append(line)
     return out
 
 

hydra-gates/scripts/lib/test_check_csrf_callers.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: EUPL-1.2
+"""Tests for check_csrf_callers (gate-48 companion).
+
+Run with:  python3 scripts/lib/test_check_csrf_callers.py
+
+Both arms, per #191's rule: arm 2 is the one that proves the companion did not
+simply declare every repository compliant. A checker that reports nothing looks
+exactly like a codebase with nothing to report.
+"""
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import check_csrf_callers as gate  # noqa: E402
+
+
+def app_with(files: dict[str, str]) -> str:
+    """Materialise a throwaway app dir; returns its path."""
+    root = tempfile.mkdtemp()
+    for rel, content in files.items():
+        path = os.path.join(root, rel)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, 'w', encoding='utf-8') as handle:
+            handle.write(content)
+    return root
+
+
+class TestCompliantCallersReportNothing(unittest.TestCase):
+    """Arm 1 — the larpingapp#298 shape: every caller already sends a token."""
+
+    def test_the_larpingapp_settings_store(self):
+        root = app_with({'src/store/modules/settings.js': """
+export default {
+    async saveSettings(config) {
+        const response = await fetch('/apps/larpingapp/api/settings', {
+            method: 'POST',
+            headers: {
+                'Content-Type': 'application/json',
+                requesttoken: OC.requestToken,
+            },
+            body: JSON.stringify(config),
+        })
+        return response.json()
+    },
+}
+"""})
+        self.assertEqual(gate.unprotected_call_sites(root), [])
+
+    def test_an_ocs_apirequest_header_counts(self):
+        root = app_with({'src/x.vue': """
+await fetch(url, { method: 'POST', headers: { 'OCS-APIREQUEST': 'true' } })
+"""})
+        self.assertEqual(gate.unprotected_call_sites(root), [])
+
+    def test_get_request_token_counts(self):
+        root = app_with({'src/x.js': """
+await fetch(url, { method: 'PUT', headers: { requesttoken: getRequestToken() } })
+"""})
+        self.assertEqual(gate.unprotected_call_sites(root), [])
+
+    def test_nextcloud_axios_injects_the_token(self):
+        root = app_with({'src/x.js': """
+import axios from '@nextcloud/axios'
+export const save = () => axios.post('/apps/x/api/settings', {})
+"""})
+        self.assertEqual(gate.unprotected_call_sites(root), [])
+
+    def test_a_plain_GET_fetch_is_not_a_mutating_call(self):
+        root = app_with({'src/x.js': "const r = await fetch('/apps/x/api/settings')\n"})
+        self.assertEqual(gate.unprotected_call_sites(root), [])
+
+    def test_an_app_with_no_src_directory(self):
+        self.assertEqual(gate.unprotected_call_sites(app_with({'lib/X.php': '<?php'})), [])
+
+
+class TestUnprotectedCallersAreReported(unittest.TestCase):
+    """Arm 2 — the gate must still catch what it was built for."""
+
+    def test_the_opencatalogi_79_delete_modal(self):
+        """The defect gate-48 exists for: a delete-modal fetch() with no CSRF
+        header. If this stopped being reported the gate would be switched off."""
+        root = app_with({'src/modals/DeleteModal.vue': """
+export default {
+    methods: {
+        async destroy(id) {
+            await fetch(`/apps/opencatalogi/api/publications/${id}`, {
+                method: 'DELETE',
+                headers: { 'Content-Type': 'application/json' },
+            })
+        },
+    },
+}
+"""})
+        found = gate.unprotected_call_sites(root)
+        self.assertEqual(len(found), 1, found)
+        self.assertIn('src/modals/DeleteModal.vue', found[0])
+        self.assertIn('DELETE', found[0])
+
+    def test_a_bare_axios_post_without_the_nextcloud_wrapper(self):
+        root = app_with({'src/x.js': """
+import axios from 'axios'
+export const save = () => axios.post('/apps/x/api/settings', {})
+"""})
+        found = gate.unprotected_call_sites(root)
+        self.assertEqual(len(found), 1, found)
+        self.assertIn('axios.post()', found[0])
+
+    def test_every_mutating_verb_is_covered(self):
+        for verb in ('POST', 'PUT', 'PATCH', 'DELETE'):
+            root = app_with({'src/x.js': f"await fetch(u, {{ method: '{verb}' }})\n"})
+            found = gate.unprotected_call_sites(root)
+            self.assertEqual(len(found), 1, f"{verb}: {found}")
+            self.assertIn(verb, found[0])
+
+    def test_a_token_on_a_DIFFERENT_call_does_not_protect_this_one(self):
+        """Paren-balanced extraction. A file-wide search would let one correct
+        call vouch for every incorrect one beside it."""
+        root = app_with({'src/x.js': """
+await fetch(a, { method: 'POST', headers: { requesttoken: OC.requestToken } })
+await fetch(b, { method: 'POST', headers: { 'Content-Type': 'application/json' } })
+"""})
+        found = gate.unprotected_call_sites(root)
+        self.assertEqual(len(found), 1, found)
+        self.assertIn(':3', found[0])
+
+    def test_node_modules_is_not_scanned(self):
+        root = app_with({
+            'src/node_modules/dep/index.js': "fetch(u, { method: 'POST' })\n",
+            'src/ok.js': "fetch(u, { method: 'POST', headers: { requesttoken: t } })\n",
+        })
+        self.assertEqual(gate.unprotected_call_sites(root), [])
+
+    def test_the_reported_line_number_is_the_call(self):
+        root = app_with({'src/x.js': "\n\n\nawait fetch(u, { method: 'POST' })\n"})
+        found = gate.unprotected_call_sites(root)
+        self.assertEqual(len(found), 1, found)
+        self.assertIn('src/x.js:4', found[0])
+
+
+class TestCli(unittest.TestCase):
+    def test_missing_argument_is_rejected(self):
+        self.assertEqual(gate.main(["check_csrf_callers.py"]), 2)
+
+    def test_extra_arguments_are_rejected(self):
+        self.assertEqual(gate.main(["check_csrf_callers.py", "a", "b"]), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()