Function
filter
v0.1.0-rc1
PII Filter
A simple filter for redacting personally identifiable information (PII)
Function ID
pii_filter
Creator
@justinrahb
Downloads
211+

Function Content
python
"""
title: Simple PII Redaction Filter
author: justinh-rahb
author_url: https://github.com/justinh-rahb
funding_url: https://github.com/open-webui
version: 0.1.0-rc1
license: MIT
"""

import re
from pydantic import BaseModel, Field
from typing import Optional


class Filter:
    class Valves(BaseModel):
        priority: int = Field(
            default=0, description="Priority level for the filter operations."
        )
        redact_email: bool = Field(default=True, description="Redact email addresses.")
        redact_phone: bool = Field(default=True, description="Redact phone numbers.")
        redact_ssn: bool = Field(
            default=True, description="Redact social security numbers."
        )
        redact_credit_card: bool = Field(
            default=True, description="Redact credit card numbers."
        )
        redact_ip_address: bool = Field(
            default=True, description="Redact IP addresses."
        )
        enabled_for_admins: bool = Field(
            default=False,
            description="Whether PII Redaction is enabled for admin users.",
        )

    def __init__(self):
        self.file_handler = False
        self.valves = self.Valves()
        self.patterns = {
            "email": re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"),
            "phone": re.compile(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b"),
            "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
            "credit_card": re.compile(r"\b(?:\d{4}[-\s]?){3}\d{4}\b"),
            "ip_address": re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"),
        }

    def redact_pii(self, text: str) -> str:
        if self.valves.redact_email:
            text = self.patterns["email"].sub("[EMAIL REDACTED]", text)
        if self.valves.redact_phone:
            text = self.patterns["phone"].sub("[PHONE REDACTED]", text)
        if self.valves.redact_ssn:
            text = self.patterns["ssn"].sub("[SSN REDACTED]", text)
        if self.valves.redact_credit_card:
            text = self.patterns["credit_card"].sub("[CREDIT CARD REDACTED]", text)
        if self.valves.redact_ip_address:
            text = self.patterns["ip_address"].sub("[IP ADDRESS REDACTED]", text)
        return text

    def inlet(self, body: dict, __user__: Optional[dict] = None) -> dict:
        print(f"inlet:{__name__}")
        print(f"inlet:body:{body}")
        print(f"inlet:user:{__user__}")

        if (
            __user__ is None
            or not __user__.get("role") == "admin"
            or self.valves.enabled_for_admins
        ):
            messages = body.get("messages", [])
            for message in messages:
                if message.get("role") == "user":
                    message["content"] = self.redact_pii(message["content"])

        return body

    def outlet(self, body: dict, __user__: Optional[dict] = None) -> dict:
        print(f"outlet:{__name__}")
        print(f"outlet:body:{body}")
        print(f"outlet:user:{__user__}")

        return body