Migrate built-in detectors to python, add file-validation detectors

Author: RobGeada

detectors/Dockerfile.builtIn

@@ -0,0 +1,26 @@
+FROM registry.access.redhat.com/ubi9/ubi-minimal as base
+RUN microdnf update -y && \
+    microdnf install -y --nodocs \
+        python-pip python-devel && \
+    pip install --upgrade --no-cache-dir pip wheel && \
+    microdnf clean all
+
+# FROM icr.io/fm-stack/ubi9-minimal-py39-torch as builder
+FROM base as builder
+
+COPY ./common/requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY ./built_in/requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+FROM builder
+
+WORKDIR /app
+ARG CACHEBUST=1
+RUN echo "$CACHEBUST"
+COPY ./common /app/detectors/common
+COPY ./built_in/* /app
+
+EXPOSE 8080
+CMD ["uvicorn", "app:app", "--workers", "4", "--host", "0.0.0.0", "--port", "8080", "--log-config", "/app/detectors/common/log_conf.yaml"]

detectors/built_in/app.py

@@ -0,0 +1,42 @@
+from fastapi import HTTPException
+
+from base_detector_registry import BaseDetectorRegistry
+from regex_detectors import RegexDetectorRegistry
+from file_type_detectors import FileTypeDetectorRegistry
+
+from prometheus_fastapi_instrumentator import Instrumentator
+from detectors.common.scheme import ContentAnalysisHttpRequest,  ContentsAnalysisResponse
+from detectors.common.app import DetectorBaseAPI as FastAPI
+
+app = FastAPI()
+Instrumentator().instrument(app).expose(app)
+
+
+registry : dict[str, BaseDetectorRegistry] = {
+    "regex": RegexDetectorRegistry(),
+    "file_type": FileTypeDetectorRegistry(),
+}
+
+@app.post("/api/v1/text/contents", response_model=ContentsAnalysisResponse)
+def detect_content(request: ContentAnalysisHttpRequest):
+    detections = []
+    for content in request.contents:
+        message_detections = []
+        for detector_kind, detector_registry in registry.items():
+            if detector_kind in request.detector_params:
+                try:
+                    message_detections += detector_registry.handle_request(content, request.detector_params)
+                except Exception as e:
+                    raise HTTPException(status_code=500, detail=str(e))
+        detections.append(message_detections)
+    return ContentsAnalysisResponse(root=detections)
+
+
+@app.get("/registry")
+def get_registry():
+    result = {}
+    for detector_type, detector_registry in registry.items():
+        result[detector_type] = {}
+        for detector_name, detector_fn in detector_registry.get_registry().items():
+            result[detector_type][detector_name] = detector_fn.__doc__
+    return result

detectors/built_in/base_detector_registry.py

@@ -0,0 +1,15 @@
+from abc import ABC, abstractmethod
+from typing import List
+
+from detectors.common.scheme import ContentAnalysisResponse
+
+class BaseDetectorRegistry(ABC):
+    def __init__(self):
+        self.registry = None
+
+    @abstractmethod
+    def handle_request(self, content: str, detector_params: dict) -> List[ContentAnalysisResponse]:
+        pass
+    
+    def get_registry(self):
+        return self.registry

detectors/built_in/file_type_detectors.py

@@ -0,0 +1,199 @@
+import json
+import jsonschema
+import xml.etree.ElementTree as ET
+import xmlschema
+import yaml
+
+from typing import List, Optional
+
+from base_detector_registry import BaseDetectorRegistry
+from detectors.common.scheme import ContentAnalysisResponse
+
+
+def is_valid_json(s: str) -> Optional[ContentAnalysisResponse]:
+    """Detect if the text contents is not valid JSON"""
+    try:
+        json.loads(s)
+        return None
+    except (ValueError, TypeError):
+        return ContentAnalysisResponse(
+            start=0,
+            end=len(s),
+            text=s,
+            detection="invalid_json",
+            detection_type= "file_type",
+            score=1.0
+        )
+
+
+def is_valid_json_schema(s: str, schema: str) -> Optional[ContentAnalysisResponse]:
+    """Detect if the text contents does not satisfy a provided JSON schema. To specify a schema, replace $SCHEMA with a JSON schema."""
+    is_valid = is_valid_json(s)
+    if is_valid is None:
+        msg_data = json.loads(s)
+    else:
+        return is_valid
+
+
+    # validate that the schema is valid json
+    try:
+        schema_data = json.loads(schema)
+    except (ValueError, TypeError):
+        return ContentAnalysisResponse(
+            start=0,
+            end=len(schema),
+            text=s,
+            detection="invalid_schema",
+            detection_type="file_type",
+            score=1.0
+        )
+
+    # validate the schema against the message
+    try:
+        jsonschema.validate(instance=msg_data, schema=schema_data)
+        return None
+    except jsonschema.ValidationError as e:
+        return ContentAnalysisResponse(
+            start=0,
+            end=len(s),
+            text=s,
+            detection="json_schema_mismatch",
+            detection_type="file_type",
+            score=1.0
+        )
+
+
+def is_valid_yaml(s: str) -> Optional[ContentAnalysisResponse]:
+    """Detect if the text contents is not valid YAML"""
+    try:
+        yaml.safe_load(s)
+        return None
+    except Exception:
+        return ContentAnalysisResponse(
+            start=0,
+            end=len(s),
+            text=s,
+            detection="invalid_yaml",
+            detection_type="file_type",
+            score=1.0,
+        )
+
+def is_valid_yaml_schema(s: str, schema) -> Optional[ContentAnalysisResponse]:
+    """Detect if the text contents does not satisfy a provided schema. To specify a schema, replace $SCHEMA with a JSON schema. That's not a typo, you validate YAML with a JSON schema!"""
+    is_valid = is_valid_yaml(s)
+    if is_valid is None:
+        msg_data = yaml.safe_load(s)
+    else:
+        return is_valid
+
+    # validate that the schema is valid json
+    try:
+        schema_data = json.loads(schema)
+    except (ValueError, TypeError):
+        return ContentAnalysisResponse(
+            start=0,
+            end=len(schema),
+            text=s,
+            detection="invalid_schema",
+            detection_type="file_type",
+            score=1.0
+        )
+
+    # validate the schema against the message
+    try:
+        jsonschema.validate(instance=msg_data, schema=schema_data)
+        return None
+    except jsonschema.ValidationError as e:
+        return ContentAnalysisResponse(
+            start=0,
+            end=len(s),
+            text=s,
+            detection="yaml_schema_mismatch",
+            detection_type="file_type",
+            score=1.0
+        )
+
+
+def is_valid_xml(s: str) -> Optional[ContentAnalysisResponse]:
+    """Detect if the text contents is not valid XML"""
+    try:
+        ET.fromstring(s)
+        return None
+    except Exception:
+        return ContentAnalysisResponse(
+            start=0,
+            end=len(s),
+            text=s,
+            detection="invalid_xml",
+            detection_type="file_type",
+            score=1.0,
+        )
+
+
+def is_valid_xml_schema(s: str, schema) -> Optional[ContentAnalysisResponse]:
+    """Detect if the text contents does not satisfy a provided XML schema. To specify a schema, replace $SCHEMA with an XML Schema Definition (XSD)"""
+    is_valid = is_valid_xml(s)
+    if is_valid is not None:
+        return is_valid
+    try:
+        # schema is expected to be a string containing the XSD
+        xs = xmlschema.XMLSchema(schema)
+    except Exception:
+        return ContentAnalysisResponse(
+            start=0,
+            end=len(schema),
+            text=s,
+            detection="invalid_xml_schema",
+            detection_type="file_type",
+            score=1.0
+        )
+
+    try:
+        xs.validate(s)
+        return None
+    except xmlschema.XMLSchemaValidationError:
+        return ContentAnalysisResponse(
+            start=0,
+            end=len(s),
+            text=s,
+            detection="xml_schema_mismatch",
+            detection_type="file_type",
+            score=1.0
+        )
+
+
+
+class FileTypeDetectorRegistry(BaseDetectorRegistry):
+    def __init__(self):
+        self.registry = {
+            "json": is_valid_json,
+            "xml": is_valid_xml,
+            "yaml": is_valid_yaml,
+            "json-with-schema:$SCHEMA": is_valid_json_schema,
+            "xml-with-schema:$SCHEMA": is_valid_xml_schema,
+            "yaml-with-schema:$SCHEMA": is_valid_yaml_schema,
+        }
+
+    def handle_request(self, content: str, detector_params: dict) -> List[ContentAnalysisResponse]:
+        detections = []
+        if "file_type" in detector_params and isinstance(detector_params["file_type"], list):
+            for file_type in detector_params["file_type"]:
+                if file_type.startswith("json-with-schema"):
+                    result = is_valid_json_schema(content, file_type.split("json-with-schema:")[1])
+                    if result is not None:
+                        detections += [result]
+                elif file_type.startswith("yaml-with-schema"):
+                    result = is_valid_yaml_schema(content, file_type.split("yaml-with-schema:")[1])
+                    if result is not None:
+                        detections += [result]
+                elif file_type.startswith("xml-with-schema"):
+                    result = is_valid_xml_schema(content, file_type.split("xml-with-schema:")[1])
+                    if result is not None:
+                        detections += [result]
+                elif file_type in self.registry:
+                    result = self.registry[file_type](content)
+                    if result is not None:
+                        detections += [result]
+                else:
+                    raise ValueError(f"Unrecognized file type: {file_type}")
+        return detections

detectors/built_in/regex_detectors.py

@@ -0,0 +1,96 @@
+import re
+from http.client import HTTPException
+from typing import List
+from base_detector_registry import BaseDetectorRegistry
+from detectors.common.scheme import ContentAnalysisResponse
+
+
+def email_address_detector(string: str) -> List[ContentAnalysisResponse]:
+    """Detect email addresses in the text contents"""
+    pattern = r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"
+    return get_regex_detections(string, pattern, "pii", "email_address")
+
+def credit_card_detector(string: str) -> List[ContentAnalysisResponse]:
+    """Detect credit cards in the text contents"""
+    pattern = r"\b(?:4\d{3}|5[0-5]\d{2}|6\d{3}|1\d{3}|3\d{3})[- ]\d{4}[- ]\d{4}[- ]\d{4}\b"
+    return get_regex_detections(string, pattern, "pii", "credit_card")
+
+
+def ipv4_detector(string: str) -> List[ContentAnalysisResponse]:
+    """Detect IPv4 addresses in the text contents"""
+    pattern = re.compile(
+        u"(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)",
+        re.IGNORECASE,
+    )
+    return get_regex_detections(string, pattern, "pii", "ipv4")
+
+def ipv6_detector(string: str) -> List[ContentAnalysisResponse]:
+    """Detect IPv6 addresses in the text contents"""
+    pattern = re.compile(
+        u"\s*(?!.*::.*::)(?:(?!:)|:(?=:))(?:[0-9a-f]{0,4}(?:(?<=::)|(?<!::):)){6}(?:[0-9a-f]{0,4}(?:(?<=::)|(?<!::):)[0-9a-f]{0,4}(?:(?<=::)|(?<!:)|(?<=:)(?<!::):)|(?:25[0-4]|2[0-4]\d|1\d\d|[1-9]?\d)(?:\.(?:25[0-4]|2[0-4]\d|1\d\d|[1-9]?\d)){3})\s*",
+        re.VERBOSE | re.IGNORECASE | re.DOTALL,
+    )
+    return get_regex_detections(string, pattern, "pii", "ipv6")
+
+# === USA Specific =================================================================================
+def ssn_detector(string: str) -> List[ContentAnalysisResponse]:
+    """Detect social security numbers in the text contents"""
+    pattern = r"\b\d{3}[- ]\d{2}[- ]\d{4}\b"
+    return get_regex_detections(string, pattern, "pii", "social_security_number")
+
+def us_phone_number_detector(string: str) -> List[ContentAnalysisResponse]:
+    """Detect US phone numbers in the text contents"""
+    pattern = r"(?:\+?1[-.\s]?)?(?:\(\d{3}\)|\d{3})[-.\s]+\d{3}[-.\s]?\d{4}\b"
+    return get_regex_detections(string, pattern, "pii", "us-phone-number")
+
+# === UK Specific =================================================================================
+def uk_post_code_detector(string: str) -> List[ContentAnalysisResponse]:
+    """Detect UK post codes in the text contents"""
+    pattern = r"\b([A-Z]{1,2}[0-9][0-9A-Z]? ?[0-9][A-Z]{2})\b"
+    return get_regex_detections(string, pattern, "pii", "uk-post-code")
+
+
+def get_regex_detections(string, pattern, detection_type, detection) -> List[ContentAnalysisResponse]:
+    detections = []
+    for match in re.finditer(pattern, string):
+        detections.append(
+            ContentAnalysisResponse(
+                start=match.start(),
+                end=match.end(),
+                text=match.string[match.start():match.end()],
+                detection_type=detection_type,
+                detection=detection,
+                score=1.0
+            ))
+    return detections
+
+# dummy function to add documention on the custom regex detector to the registr
+def custom_regex_documenter():
+    """Replace $CUSTOM_REGEX with a custom regex to define your own regex detector"""
+
+
+# === ROUTER =======================================================================================
+class RegexDetectorRegistry(BaseDetectorRegistry):
+    def __init__(self):
+        self.registry =  {
+            "credit-card": credit_card_detector,
+            "email": email_address_detector,
+            "ipv4": ipv4_detector,
+            "ipv6": ipv6_detector,
+            "us-phone-number": us_phone_number_detector,
+            "us-social-security-number": ssn_detector,
+            "uk-post-code": uk_post_code_detector,
+            "$CUSTOM_REGEX": custom_regex_documenter,
+        }
+
+    def handle_request(self, content: str, detector_params: dict) -> List[ContentAnalysisResponse]:
+        detections = []
+        if "regex" in detector_params and isinstance(detector_params["regex"], list):
+            for regex in detector_params["regex"]:
+                if regex == "$CUSTOM_REGEX":
+                    pass
+                elif regex in self.registry:
+                    detections += self.registry[regex](content)
+                else:
+                    detections += get_regex_detections(content, regex, "regex", "custom-regex")
+        return detections

detectors/built_in/requirements.txt

@@ -0,0 +1,3 @@
+markdown==3.8.2
+jsonschema==4.24.0
+xmlschema==4.1.0

detectors/common/app.py

@@ -107,7 +107,7 @@ def main(app):
     config = {
         "server": {
             "host": "0.0.0.0",
-            "port": "8000",
+            "port": "8080",
             "workers": 1,
             "limit_concurrency": 1000,
             "timeout_keep_alive": 30,