v0.4.1: schema validation, URL hardening, secrets patterns, tests

Author: taladari

.gitignore

@@ -77,3 +77,8 @@ audit.jsonl
 
 # Example artifacts
 examples/demo/
+
+
+STRATEGY.md
+CORE_DEV_PLAN.md
+DISTRIBUTION.md

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## [0.4.1] - 2025-09-02
+### Added
+- Schema validation (optional, via jsonschema)
+- URL scanner hardening (IP/punycode)
+- Secrets patterns extended
+- Tests
+
+
 ## [0.4.0] - 2025-08-30
 ### Added
 - GraphRAG support:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "rag-firewall"
-version = "0.4.0"
+version = "0.4.1"
 description = "Client-side retrieval firewall for RAG systems — blocks prompt injection and secret leaks, re-ranks stale or untrusted content, and keeps all data inside your environment."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.9"
@@ -24,9 +24,13 @@ dev = [
   "networkx>=3.2"
 ]
 
+[tool.setuptools.package-data]
+"rag_firewall" = ["schema/*.json"]
+
 [project.urls]
 Homepage = "https://github.com/taladari/rag-firewall"
 Repository = "https://github.com/taladari/rag-firewall"
 
 [project.scripts]
 ragfw = "rag_firewall.cli:main"
+

rag_firewall/firewall.py

@@ -18,6 +18,26 @@ def from_yaml(cls, path):
         with open(path,"r",encoding="utf-8") as f:
             if not yaml: raise RuntimeError("PyYAML is required to load YAML configs.")
             cfg=yaml.safe_load(f)
+        # Optional JSON Schema validation if jsonschema is available
+        try:
+            import json
+            from importlib.resources import files
+            import jsonschema  # type: ignore
+            schema_path = files("rag_firewall").joinpath("schema/firewall.schema.json")
+            with open(schema_path, "r", encoding="utf-8") as sf:
+                schema = json.load(sf)
+            jsonschema.validate(instance=cfg, schema=schema)
+        except ModuleNotFoundError:
+            # jsonschema not installed; skip validation gracefully
+            pass
+        except Exception as ve:
+            # If validation fails, surface a clear error
+            try:
+                import jsonschema  # noqa: F401
+                raise ValueError(f"Invalid firewall config: {getattr(ve, 'message', str(ve))}") from ve
+            except Exception:
+                # Unknown error during validation; continue without blocking
+                pass
         scanners=[]
         from .scanners.regex_scanner import RegexInjectionScanner
         from .scanners.pii_scanner import PIIScanner

rag_firewall/scanners/secrets_scanner.py

@@ -7,7 +7,13 @@
 (r"ghp_[A-Za-z0-9]{36}","github_token"),(r"AIza[0-9A-Za-z\-_]{35}","google_api_key"),
 (r"xox[abp]-\d{10,}-\d{10,}-[A-Za-z0-9-]{24,}","slack_token"),(r"sk-[A-Za-z0-9]{32,}","generic_sk_token"),
 (r"(?i)bearer\s+[A-Za-z0-9\-_\.=]{20,}","bearer_token"),
-(r"-----BEGIN (?:RSA|OPENSSH|EC) PRIVATE KEY-----","private_key")]
+(r"-----BEGIN (?:RSA|OPENSSH|EC) PRIVATE KEY-----","private_key"),
+# Additional high-signal patterns
+(r"hf_[A-Za-z0-9]{30,}","huggingface_token"),
+(r"dapi[a-zA-Z0-9]{24}","databricks_token"),
+(r"https://hooks\.slack\.com/services/[A-Za-z0-9/+]{20,}","slack_webhook"),
+(r"(?i)azure[_\s-]key[vV]ault|AZURE_[A-Z0-9_]{8,}","azure_secret_suspect"),
+(r"secret_[A-Za-z0-9]{32,}","generic_secret_token")]
 class SecretsScanner:
     def __init__(self, extra_patterns=None):
         import regex as re

rag_firewall/scanners/url_scanner.py

@@ -3,6 +3,7 @@
 
 import regex as re
 from urllib.parse import urlparse
+import ipaddress
 URL_RE=re.compile(r"https?://[\w\-\.:%#@/\?=~\+,&]+", re.I)
 class URLScanner:
     def __init__(self, allowlist=None, denylist=None):
@@ -13,6 +14,17 @@ def scan(self, text, metadata):
         for m in URL_RE.findall(t):
             host=(urlparse(m).hostname or "").lower()
             sev="low"; reason="url_found"
+            # Flag IP literals (IPv4/IPv6)
+            try:
+                if host:
+                    ipaddress.ip_address(host)
+                    out.append({"scanner":"url","match":host,"severity":"high","reason":"ip_literal"})
+                    # Do not continue; also evaluate allow/deny checks below
+            except ValueError:
+                pass
+            # Flag punycode domains
+            if host.startswith("xn--"):
+                out.append({"scanner":"url","match":host,"severity":"high","reason":"punycode_host"})
             if self.denylist and any(host==d or host.endswith("."+d) for d in self.denylist):
                 sev="high"; reason="denylist_domain"
             elif self.allowlist and not any(host==d or host.endswith("."+d) for d in self.allowlist):

rag_firewall/schema/__init__.py

@@ -0,0 +1,49 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "title": "RAG Firewall Config v1",
+  "type": "object",
+  "properties": {
+    "scanners": {
+      "type": "array",
+      "items": {
+        "type": "object",
+        "properties": {
+          "type": { "type": "string", "enum": ["regex_injection","pii","secrets","encoded","url","conflict"] },
+          "patterns": { "type": "array", "items": { "type": "string" } },
+          "extra_patterns": { "type": "array", "items": { "type": "string" } },
+          "min_len": { "type": "integer", "minimum": 1 },
+          "ratio_threshold": { "type": "number", "minimum": 0, "maximum": 1 },
+          "allowlist": { "type": "array", "items": { "type": "string" } },
+          "denylist": { "type": "array", "items": { "type": "string" } },
+          "stale_days": { "type": "integer", "minimum": 1 }
+        },
+        "required": ["type"],
+        "additionalProperties": true
+      }
+    },
+    "policies": {
+      "type": "array",
+      "items": {
+        "type": "object",
+        "properties": {
+          "name": { "type": "string" },
+          "action": { "type": "string", "enum": ["allow","deny","rerank"] },
+          "match": { "type": "object" },
+          "weight": {
+            "type": "object",
+            "properties": {
+              "recency": { "type": "number" },
+              "relevance": { "type": "number" },
+              "provenance": { "type": "number" }
+            },
+            "additionalProperties": false
+          }
+        },
+        "required": ["name","action"],
+        "additionalProperties": true
+      }
+    }
+  },
+  "required": ["scanners","policies"],
+  "additionalProperties": false
+}

rag_firewall/schema/firewall.schema.json

@@ -0,0 +1,50 @@
+import os, json, tempfile
+import pytest
+
+from rag_firewall.firewall import Firewall
+
+VALID_CFG = {
+    "scanners": [
+        {"type": "regex_injection"},
+        {"type": "secrets"},
+        {"type": "url", "allowlist": ["good.example.com"], "denylist": ["evil.example.com"]},
+        {"type": "conflict", "stale_days": 120}
+    ],
+    "policies": [
+        {"name": "allow_default", "action": "allow"}
+    ]
+}
+
+INVALID_ACTION_CFG = {
+    "scanners": [ {"type": "regex_injection"} ],
+    "policies": [ {"name": "bad", "action": "block"} ]  # not in enum
+}
+
+INVALID_SCANNER_TYPE_CFG = {
+    "scanners": [ {"type": "unknown_scanner"} ],
+    "policies": [ {"name": "allow_default", "action": "allow"} ]
+}
+
+def _write_tmp_yaml(obj):
+    import yaml
+    fd, path = tempfile.mkstemp(suffix=".yaml")
+    with os.fdopen(fd, "w", encoding="utf-8") as f:
+        yaml.safe_dump(obj, f)
+    return path
+
+@pytest.mark.parametrize("cfg", [VALID_CFG])
+def test_valid_config_passes_validation(cfg):
+    path = _write_tmp_yaml(cfg)
+    # Should not raise; might skip validation if jsonschema not installed
+    fw = Firewall.from_yaml(path)
+    assert isinstance(fw, Firewall)
+
+@pytest.mark.parametrize("cfg", [INVALID_ACTION_CFG, INVALID_SCANNER_TYPE_CFG])
+def test_invalid_config_raises_when_jsonschema_present(cfg):
+    try:
+        import jsonschema  # noqa: F401
+    except Exception:
+        pytest.skip("jsonschema not installed; validation is optional")
+    path = _write_tmp_yaml(cfg)
+    with pytest.raises(ValueError):
+        Firewall.from_yaml(path)

tests/test_config_validation.py

@@ -0,0 +1,20 @@
+from rag_firewall.scanners.secrets_scanner import SecretsScanner
+
+
+def test_huggingface_and_databricks_and_slack_webhook_detected():
+    s = SecretsScanner()
+    text = "hf_abcdefghijklmnopqrstuvwxyzABCDE dapiABCDEFGHIJKLMNOPQRSTUVWX https://hooks.slack.com/services/T12345/A12345/ABCDEFghijklmnop"
+    findings = s.scan(text, {})
+    names = {f.get("match") for f in findings}
+    assert "huggingface_token" in names
+    assert "databricks_token" in names
+    assert "slack_webhook" in names
+
+
+def test_azure_like_and_generic_secret_detected():
+    s = SecretsScanner()
+    text = "AZURE_SECRET_ABCDEFGH secret_abcdefghijklmnopqrstuvwxyz123456"
+    findings = s.scan(text, {})
+    names = {f.get("match") for f in findings}
+    assert "azure_secret_suspect" in names or any("azure" in n for n in names)
+    assert "generic_secret_token" in names