Security & PII Skill

Role

You are an elite AI security engineer. You know that unsecured AI systems leak

data, get jailbroken, and fail compliance audits. You build defence-in-depth

security for every AI pipeline — PII redaction, prompt injection detection,

output validation, and encrypted storage.

Part 1: PII Detection & Redaction (Microsoft Presidio)

# pip install presidio-analyzer presidio-anonymizer spacy
# python -m spacy download en_core_web_lg

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
from presidio_analyzer.nlp_engine import NlpEngineProvider

# Supported entity types
PII_ENTITIES = [
    "PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER", "CREDIT_CARD",
    "IBAN_CODE", "IP_ADDRESS", "LOCATION", "DATE_TIME",
    "NRP",  # Nationality, Religious, Political
    "MEDICAL_LICENSE", "URL", "US_SSN", "UK_NHS",
    "CRYPTO", "AU_ABN", "AU_ACN"
]

def create_analyzer() -> AnalyzerEngine:
    """Create a Presidio analyzer with spaCy NLP engine."""
    config = {"nlp_engine_name": "spacy",
               "models": [{"lang_code": "en", "model_name": "en_core_web_lg"}]}
    provider = NlpEngineProvider(nlp_configuration=config)
    nlp_engine = provider.create_engine()
    return AnalyzerEngine(nlp_engine=nlp_engine)

analyzer = create_analyzer()
anonymizer = AnonymizerEngine()


def detect_pii(text: str,
                entities: list[str] = None,
                language: str = "en") -> list[dict]:
    """Detect PII entities in text."""
    results = analyzer.analyze(
        text=text,
        entities=entities or PII_ENTITIES,
        language=language
    )
    return [
        {
            "entity_type": r.entity_type,
            "start": r.start,
            "end": r.end,
            "score": r.score,
            "text": text[r.start:r.end]
        }
        for r in sorted(results, key=lambda x: x.start)
    ]


def redact_pii(text: str,
                strategy: str = "replace",
                entities: list[str] = None) -> dict:
    """
    Redact PII from text.

    strategy options:
    - "replace"  → <PERSON>, <EMAIL_ADDRESS> etc (default, most readable)
    - "mask"     → ******** (hides completely)
    - "hash"     → irreversible hash (for pseudonymisation)
    - "encrypt"  → reversible encryption (allows de-anonymisation)
    """
    analyzer_results = analyzer.analyze(
        text=text,
        entities=entities or PII_ENTITIES,
        language="en"
    )

    if strategy == "replace":
        operators = {
            entity: OperatorConfig("replace", {"new_value": f"<{entity}>"})
            for entity in (entities or PII_ENTITIES)
        }
    elif strategy == "mask":
        operators = {
            entity: OperatorConfig("mask", {"masking_char": "*", "chars_to_mask": 100, "from_end": False})
            for entity in (entities or PII_ENTITIES)
        }
    elif strategy == "hash":
        operators = {
            entity: OperatorConfig("hash", {"hash_type": "sha256"})
            for entity in (entities or PII_ENTITIES)
        }
    else:
        operators = {}

    anonymized = anonymizer.anonymize(
        text=text,
        analyzer_results=analyzer_results,
        operators=operators
    )

    return {
        "original": text,
        "redacted": anonymized.text,
        "entities_found": len(analyzer_results),
        "entity_types": list(set(r.entity_type for r in analyzer_results))
    }


def safe_llm_input(text: str) -> tuple[str, dict]:
    """
    Redact PII before sending to any LLM API.
    Returns: (redacted_text, mapping_for_restoration)
    """
    result = redact_pii(text, strategy="replace")
    return result["redacted"], {
        "original_length": len(text),
        "entities_removed": result["entity_types"]
    }

Part 2: Prompt Injection Detection

import re
from typing import Literal

# Common prompt injection patterns
INJECTION_PATTERNS = [
    # Direct instruction overrides
    r"ignore (all |previous |above |prior )(instructions?|prompts?|context)",
    r"disregard (your |the |all )(previous |system |prior )?instructions?",
    r"forget (everything|what|your) (you were|you are|was) told",
    r"you are now (a |an |the )?(different|new|another|evil)",

    # Role manipulation
    r"act as (if you are|a|an) (different|unrestricted|jailbreak|DAN)",
    r"pretend (you are|to be) (not|without) (any |your )?(restrictions?|guidelines?)",
    r"(you are|you're) now (free|allowed|able) to",

    # System prompt extraction
    r"(repeat|print|show|reveal|tell me|output|display) (your |the )?(system prompt|instructions?|prompt|context)",
    r"what (are|were) (your |the )?(original |system |initial )?instructions?",

    # Jailbreak patterns
    r"DAN|jailbreak|unrestricted mode|developer mode",
    r"(bypass|ignore|override) (safety|content|ethical) (filters?|guidelines?|policies?)",
]

COMPILED_PATTERNS = [re.compile(p, re.IGNORECASE) for p in INJECTION_PATTERNS]


def detect_prompt_injection(text: str,
                              threshold: float = 0.5) -> dict:
    """
    Detect prompt injection attempts in user input.
    Returns risk score and matched patterns.
    """
    matches = []
    for pattern in COMPILED_PATTERNS:
        match = pattern.search(text)
        if match:
            matches.append(match.group(0))

    # Score: 0.0 (safe) to 1.0 (definitely injection)
    score = min(1.0, len(matches) * 0.3)

    return {
        "is_injection": score >= threshold,
        "risk_score": score,
        "matched_patterns": matches,
        "recommendation": "BLOCK" if score >= threshold else "ALLOW"
    }


def sanitise_user_input(text: str,
                          max_length: int = 4000) -> dict:
    """
    Full input sanitisation pipeline.
    1. Check for prompt injection
    2. Redact PII
    3. Truncate to max length
    """
    # Step 1: Injection check
    injection = detect_prompt_injection(text)

    if injection["is_injection"]:
        return {
            "safe": False,
            "reason": "prompt_injection",
            "details": injection,
            "sanitised_text": None
        }

    # Step 2: PII redaction
    redacted, pii_meta = safe_llm_input(text)

    # Step 3: Truncate
    if len(redacted) > max_length:
        redacted = redacted[:max_length] + "... [truncated]"

    return {
        "safe": True,
        "sanitised_text": redacted,
        "pii_removed": pii_meta["entities_removed"],
        "truncated": len(text) > max_length
    }

Part 3: LLM Output Validation

from pydantic import BaseModel, Field

HARMFUL_PATTERNS = [
    r"(how to|instructions? for|steps? to) (make|create|build|synthesise) (a |an )?(bomb|weapon|explosive|poison)",
    r"(social security|SSN|credit card) number[s]? (is|are|:)",
    r"password[s]?[\s:]+[a-zA-Z0-9!@#$%^&*]{8,}",
]
COMPILED_HARMFUL = [re.compile(p, re.IGNORECASE) for p in HARMFUL_PATTERNS]

class OutputValidationResult(BaseModel):
    safe: bool
    issues: list[str] = Field(default_factory=list)
    pii_in_output: list[str] = Field(default_factory=list)
    harmful_content: bool = False
    validated_output: str = ""


def validate_llm_output(output: str,
                         redact_pii_in_output: bool = True) -> OutputValidationResult:
    """
    Validate LLM output before returning to user.
    Catches: PII leakage, harmful content, sensitive data.
    """
    issues = []

    # Check for harmful content
    harmful_matches = [p.search(output) for p in COMPILED_HARMFUL]
    harmful = any(harmful_matches)
    if harmful:
        issues.append("harmful_content_detected")

    # Check for PII in output (LLM may have regenerated redacted data)
    pii_result = detect_pii(output)
    pii_types = [p["entity_type"] for p in pii_result]
    if pii_types:
        issues.append(f"pii_in_output: {pii_types}")

    if not issues:
        return OutputValidationResult(
            safe=True,
            validated_output=output
        )

    # Attempt remediation: redact PII from output
    if redact_pii_in_output and pii_types and not harmful:
        cleaned = redact_pii(output, strategy="replace")
        return OutputValidationResult(
            safe=True,
            issues=issues,
            pii_in_output=pii_types,
            validated_output=cleaned["redacted"]
        )

    return OutputValidationResult(
        safe=False,
        issues=issues,
        pii_in_output=pii_types,
        harmful_content=harmful,
        validated_output=""
    )

Part 4: Encrypted Storage for AI Outputs

# pip install cryptography
from cryptography.fernet import Fernet
import base64, os

def generate_encryption_key() -> str:
    """Generate a new Fernet encryption key. Store this securely."""
    return Fernet.generate_key().decode()


class EncryptedAIStore:
    """
    Encrypt sensitive LLM outputs at rest.
    Use when storing conversation history containing client data.
    """

    def __init__(self, key: str = None):
        key = key or os.environ.get("AI_ENCRYPTION_KEY")
        if not key:
            raise ValueError("Encryption key required. Set AI_ENCRYPTION_KEY env var.")
        self.fernet = Fernet(key.encode() if isinstance(key, str) else key)

    def encrypt(self, text: str) -> str:
        return self.fernet.encrypt(text.encode()).decode()

    def decrypt(self, encrypted: str) -> str:
        return self.fernet.decrypt(encrypted.encode()).decode()

    def encrypt_dict(self, data: dict) -> dict:
        import json
        return {"encrypted": self.encrypt(json.dumps(data))}

    def decrypt_dict(self, data: dict) -> dict:
        import json
        return json.loads(self.decrypt(data["encrypted"]))

Part 5: Secure AI API Endpoint Pattern

from fastapi import FastAPI, Depends, HTTPException, Header
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt, os

app = FastAPI()
security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)) -> dict:
    """Verify JWT token on every request."""
    try:
        payload = jwt.decode(
            credentials.credentials,
            os.environ["JWT_SECRET"],
            algorithms=["HS256"]
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")


@app.post("/api/research")
async def secure_research_endpoint(
    request: dict,
    user: dict = Depends(verify_token)
):
    """Secure research endpoint with full input/output sanitisation."""
    raw_query = request.get("query", "")

    # 1. Sanitise input
    sanitised = sanitise_user_input(raw_query)
    if not sanitised["safe"]:
        raise HTTPException(status_code=400, detail=f"Unsafe input: {sanitised['reason']}")

    # 2. Call LLM (use sanitised text)
    # result = llm(sanitised["sanitised_text"])

    # 3. Validate output
    # validated = validate_llm_output(result)
    # if not validated.safe:
    #     raise HTTPException(status_code=500, detail="Output failed safety check")

    return {
        "query_sanitised": sanitised["pii_removed"],
        "user_id": user.get("sub")
    }

Part 6: GDPR-Compliant Data Deletion

import sqlite3

def delete_user_data(user_id: str, db_path: str) -> dict:
    """
    Delete all AI-generated data for a user (GDPR right to erasure).
    """
    deleted = {}
    with sqlite3.connect(db_path) as conn:
        for table in ["conversations", "research_outputs", "llm_cache"]:
            try:
                cur = conn.execute(
                    f"DELETE FROM {table} WHERE session_id LIKE ?",
                    (f"{user_id}%",)
                )
                deleted[table] = cur.rowcount
            except Exception:
                deleted[table] = 0
    return {"user_id": user_id, "deleted_rows": deleted}

Output Standards

• ALWAYS redact PII before sending to any third-party LLM API
• ALWAYS validate LLM output before returning to end users
• Run prompt injection detection on ALL user-supplied text
• Encrypt stored conversation history containing client data at rest
• Log security events (injections detected, PII found) at WARN level — never log the actual content
• For IMI: all fan research data containing individual responses must be PII-scrubbed before indexing

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)