Move OWASP scanner into scanner/ subdirectory + add workflow

.github/workflows/scan.yml

@@ -0,0 +1,82 @@
+name: OWASP PR Scanner
+
+on:
+  pull_request:
+    paths:
+      - 'src/**'
+      - 'backend/**'
+      - 'app/**'
+      - 'services/**'
+      - '.github/workflows/**'
+      - 'scanner/**'
+
+jobs:
+  scan:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      - name: Install deps
+        run: |
+          python -m pip install -U pip
+          if [ -f scanner/requirements.txt ]; then
+            pip install -r scanner/requirements.txt
+          elif [ -f requirements.txt ]; then
+            pip install -r requirements.txt
+          fi
+
+      - name: Determine changed files for this PR
+        id: diff
+        shell: bash
+        run: |
+          BASE_SHA="${{ github.event.pull_request.base.sha }}"
+          HEAD_SHA="${{ github.event.pull_request.head.sha }}"
+
+          RAW=$(git diff --name-only "$BASE_SHA" "$HEAD_SHA" || true)
+
+          APP_CHANGED=$(echo "$RAW" \
+            | grep -E '\.(js|jsx|ts|tsx|py|java|go|rb|php|html|css|md)$' \
+            | grep -E '^(src/|backend/|app/|services/)' || true)
+
+          SCANNER_ONLY=$(echo "$RAW" | grep -E '^scanner/' || true)
+
+          if [ -z "$APP_CHANGED" ] && [ -n "$SCANNER_ONLY" ]; then
+            echo "only_scanner_changes=true" >> $GITHUB_OUTPUT
+            exit 0
+          fi
+
+          if [ -z "$APP_CHANGED" ]; then
+            APP_CHANGED="$(git ls-files src backend app services 2>/dev/null || true)"
+          fi
+
+          echo "CHANGED_FILES<<EOF" >> "$GITHUB_ENV"
+          echo "$APP_CHANGED" >> "$GITHUB_ENV"
+          echo "EOF" >> "$GITHUB_ENV"
+
+      - name: Skip when only scanner/** changed
+        if: steps.diff.outputs.only_scanner_changes == 'true'
+        run: echo "Only scanner/** changed; skipping scan."
+
+      - name: Run OWASP scanner on changed files
+        if: steps.diff.outputs.only_scanner_changes != 'true'
+        shell: bash
+        run: |
+          if [ -z "${CHANGED_FILES}" ]; then
+            echo "Nothing to scan."
+            exit 0
+          fi
+          EXIT=0
+          while IFS= read -r file; do
+            [ -z "$file" ] && continue
+            echo "Scanning: $file"
+            python -m scanner.main "$file" || EXIT=1
+          done <<< "${CHANGED_FILES}"
+          exit $EXIT

.gitignore

@@ -0,0 +1 @@
+"scanner/venv/" 

scanner/core.py

@@ -0,0 +1,146 @@
+# Responsibilities:
+# - Reads target file, stores code lines
+# - Manages vulnerability list
+# - Runs all rule checks (auto-discovers rules in scanner/rules)
+# - Provides add_vulnerability callback
+# - Prints a grouped, colourised report
+
+import os
+import importlib
+import pkgutil
+import scanner.rules as rules_pkg
+
+
+# -------- Rule auto-discovery --------
+def _load_rule_modules():
+    modules = []
+    for _, modname, _ in pkgutil.iter_modules(rules_pkg.__path__):
+        if modname.startswith("_"):
+            continue  # skip __init__, _template, etc.
+        mod = importlib.import_module(f"{rules_pkg.__name__}.{modname}")
+        if hasattr(mod, "check"):
+            modules.append(mod)
+
+    # Stable order: by CATEGORY "A01: ..." if provided, else by module name
+    def key(m):
+        cat = getattr(m, "CATEGORY", "")
+        head = cat.split(":", 1)[0].strip() if cat else ""
+        return (0, int(head[1:])) if head.startswith("A") and head[1:].isdigit() else (1, m.__name__)
+
+    return sorted(modules, key=key)
+
+
+RULE_MODULES = _load_rule_modules()
+
+
+# -------- Scanner --------
+class VulnerabilityScanner:
+    def __init__(self, file_path):
+        self.file_path = file_path
+        self.code_lines = []
+        self.vulnerabilities = []
+
+    def add_vulnerability(self, category, description, line, severity, confidence):
+        self.vulnerabilities.append(
+            {
+                "category": category,
+                "description": description,
+                "line": line,
+                "severity": severity,
+                "confidence": confidence,
+            }
+        )
+
+    def parse_file(self):
+        if not os.path.exists(self.file_path):
+            print(f"File {self.file_path} does not exist.")
+            return False
+        with open(self.file_path, "r", encoding="utf-8") as f:
+            self.code_lines = f.readlines()
+        return True
+
+    def run_checks(self):
+        for rule in RULE_MODULES:
+            # each rule exposes: check(code_lines, add_vulnerability)
+            rule.check(self.code_lines, self.add_vulnerability)
+
+    def run(self):
+        if not self.parse_file():
+            return
+        self.run_checks()
+
+    def report(self):
+        # ---- colour helpers ----
+        def supports_truecolor() -> bool:
+            return os.environ.get("COLORTERM", "").lower() in ("truecolor", "24bit")
+
+        def rgb(r, g, b) -> str:
+            return f"\033[38;2;{r};{g};{b}m"
+
+        ANSI = {
+            "reset": "\033[0m",
+            "bold": "\033[1m",
+            "cyan": "\033[96m",
+            "magenta": "\033[95m",
+            "yellow": "\033[93m",
+            "red": "\033[91m",
+            "green": "\033[92m",
+            "blue": "\033[94m",
+        }
+
+        TRUECOLOR = supports_truecolor()
+
+        # Severity colours (true-color -> fallback)
+        CRIT = (rgb(220, 20, 60) if TRUECOLOR else ANSI["red"] + ANSI["bold"])  # crimson
+        HIGH = (rgb(255, 0, 0) if TRUECOLOR else ANSI["red"])                   # red
+        MED = (rgb(255, 165, 0) if TRUECOLOR else ANSI["yellow"])               # orange-ish
+        LOW = (rgb(0, 200, 0) if TRUECOLOR else ANSI["green"])                  # green
+
+        RESET = ANSI["reset"]
+        BOLD = ANSI["bold"]
+        HDR = (rgb(180, 130, 255) if TRUECOLOR else ANSI["magenta"])            # section header
+        TITLE = (rgb(120, 220, 200) if TRUECOLOR else ANSI["cyan"])             # title
+        SUM = (rgb(255, 215, 0) if TRUECOLOR else ANSI["yellow"])               # summary label
+
+        sev_color = {"CRITICAL": CRIT, "HIGH": HIGH, "MEDIUM": MED, "LOW": LOW}
+
+        print(f"\n{BOLD}{TITLE}Scan Results for {self.file_path}:{RESET}")
+
+        if not self.vulnerabilities:
+            ok = rgb(0, 200, 0) if TRUECOLOR else ANSI["green"]
+            print(f"{ok}✅ No vulnerabilities found.{RESET}")
+            return
+
+        # Group by category
+        groups = {}
+        for v in self.vulnerabilities:
+            groups.setdefault(v["category"], []).append(v)
+
+        def cat_key(cat: str):
+            head = cat.split(":", 1)[0].strip()
+            return (0, int(head[1:])) if head.startswith("A") and head[1:].isdigit() else (1, cat.lower())
+
+        for cat in sorted(groups.keys(), key=cat_key):
+            items = sorted(groups[cat], key=lambda x: x["line"])
+            # tally
+            sev_counts = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0}
+            for v in items:
+                sev_counts[v["severity"]] = sev_counts.get(v["severity"], 0) + 1
+
+            total = len(items)
+            print(f"\n{BOLD}{HDR}=== {cat} ({total} finding{'s' if total != 1 else ''}) ==={RESET}")
+
+            chips = []
+            for k in ["CRITICAL", "HIGH", "MEDIUM", "LOW"]:
+                n = sev_counts.get(k, 0)
+                if n:
+                    chips.append(f"{sev_color[k]}{k.title()}{RESET}: {n}")
+            if chips:
+                print(f"{SUM}Summary:{RESET} " + ", ".join(chips))
+
+            for v in items:
+                sc = sev_color.get(v["severity"], ANSI["blue"])
+                print(f"\n  {BOLD}• Line {v['line']} |{RESET} "
+                      f"Severity {sc}{v['severity']}{RESET} | "
+                      f"Confidence {v['confidence']}")
+                print(f"    → {v['description']}")

scanner/main.py

@@ -0,0 +1,21 @@
+# Entry point for the OWASP PR Scanner CLI tool.
+# This script parses the command-line arguments (i.e., the file path to scan),
+# initializes the VulnerabilityScanner with the specified file, runs all rule checks,
+# and prints a formatted vulnerability report to the console.
+
+
+import argparse
+from .core import VulnerabilityScanner
+
+
+def main():
+    parser = argparse.ArgumentParser(description="OWASP PR Vulnerability Scanner")
+    parser.add_argument("path", help="Path to Python file to scan")
+    args = parser.parse_args()
+
+    scanner = VulnerabilityScanner(args.path)
+    scanner.run()
+    scanner.report()
+
+if __name__ == "__main__":
+    main()

scanner/rules/_template.py

@@ -0,0 +1,11 @@
+# Template for adding new OWASP rule modules
+def check(code_lines, add_vulnerability):
+    for i, line in enumerate(code_lines):
+        if "pattern" in line:  # replace with real logic
+            add_vulnerability(
+                "Axx: Rule Name",
+                f"Description: {line.strip()}",
+                i + 1,
+                "HIGH",
+                "MEDIUM"
+            )

scanner/rules/auth_failures.py

@@ -0,0 +1,45 @@
+# A07:2021 – Identification and Authentication Failures
+# Detects default credentials (`admin`, `password`)  
+# Flags login routes without auth checks   
+# Warns about disabled TLS verification (`verify=False`) 
+import re
+
+def check(code_lines, add_vulnerability):
+    for i, line in enumerate(code_lines):
+        # Flask/Django style routes that should require auth
+        if re.search(r"@app\.route\([\"'](/login|/auth|/signin)[\"']", line):
+            add_vulnerability(
+                "A07: Identification and Authentication Failures",
+                f"Authentication-related route without explicit auth checks: {line.strip()}",
+                i + 1,
+                "HIGH",
+                "MEDIUM"
+            )
+
+        # Python requests with TLS verify disabled
+        if "requests." in line and "verify=False" in line:
+            add_vulnerability(
+                "A07: Identification and Authentication Failures",
+                f"Insecure TLS verification disabled: {line.strip()}",
+                i + 1,
+                "HIGH",
+                "HIGH"
+            )
+
+        # Hardcoded default creds
+        if re.search(r"(user(name)?\s*=\s*['\"](admin|root)['\"])", line, re.IGNORECASE):
+            add_vulnerability(
+                "A07: Identification and Authentication Failures",
+                f"Hardcoded default username detected: {line.strip()}",
+                i + 1,
+                "HIGH",
+                "HIGH"
+            )
+        if re.search(r"(password\s*=\s*['\"](admin|1234|password)['\"])", line, re.IGNORECASE):
+            add_vulnerability(
+                "A07: Identification and Authentication Failures",
+                f"Hardcoded default password detected: {line.strip()}",
+                i + 1,
+                "HIGH",
+                "HIGH"
+            )