""" Workplace-Pulse-Telemetry: Local-First Zero Data Leakage Anonymizer Engineered by Fabio Torres (neurodeveloper11) Performs deterministic local PII detection, redaction, and salted pseudonymization guaranteeing that no confidential personal data leaves the local execution boundary. """ import re import hashlib from typing import List, Tuple, Dict, Optional from src.schemas import RawMessage, SanitizedMessage, PIIEntity class LocalAnonymizer: """ On-premise zero-leakage anonymization engine. Detects and redacts emails, phone numbers, national identification numbers (CC/DNI/SSN), financial data (salaries, credit cards), network addresses, and personal mentions. """ def __init__(self, salt: str = "workplace_pulse_default_salt_2026"): self.salt = salt self._user_pseudonym_cache: Dict[str, str] = {} # 1. Email Regex self.email_pattern = re.compile( r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b" ) # 2. Phone Numbers (Colombia, USA, International formats) self.phone_pattern = re.compile( r"(?:\+?57\s*)?(?:\b3\d{2}[\s.-]?\d{3}[\s.-]?\d{4}\b|\b(?:\+?1[\s.-]?)?\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}\b)" ) # 3. National IDs (Cédula de Ciudadanía, DNI, SSN) self.id_pattern = re.compile( r"\b(?:CC|C\.C\.|Cédula|Cedula|DNI|Identificación|ID)[\s#:.]*([0-9]{1,3}(?:\.[0-9]{3}){2}|[0-9]{6,10})\b", re.IGNORECASE ) self.ssn_pattern = re.compile(r"\b\d{3}-\d{2}-\d{4}\b") # 4. Financial & Salary Data (e.g. $12.000.000 COP, $4,500 USD, 4500 USD, salario de 5 millones) self.financial_pattern = re.compile( r"(?:(?:\$|COP|USD|EUR)\s*[\d]{1,3}(?:[.,]\d{3})*(?:[.,]\d{2})?|\b\d+(?:[.,]\d+)*\s*(?:pesos|dólares|dolares|COP|USD|EUR)\b|\b(?:sueldo|salario|pago|honorarios|bono)\s+(?:de\s+)?[\$]?[\d.,]+(?:\s*(?:millones|mil|k|USD|COP))?\b|\b(?:\d{4}[ -]?){3}\d{4}\b)", re.IGNORECASE ) # 5. IP Addresses & Internal Domain hosts self.ip_pattern = re.compile( r"\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b" ) # 6. Corporate Chat Mentions (@carlos, @juan.perez) self.mention_pattern = re.compile(r"@([a-zA-Z0-9._-]+)") # 7. Common Spanish/English Greetings with Proper Names (e.g. "Hola Carlos,", "Buenos días María,") self.greeting_name_pattern = re.compile( r"\b(Hola|Buenos días|Buenas tardes|Estimado|Estimada|Dear|Hi|Hello)\s+([A-ZÁÉÍÓÚÑ][a-záéíóúñ]+(?:\s+[A-ZÁÉÍÓÚÑ][a-záéíóúñ]+)?)(?=[,\s!?\n]|$)", re.IGNORECASE ) def get_pseudonym(self, raw_id: str) -> str: """Generates a deterministic, collision-resistant pseudonym salted with SHA-256.""" if raw_id not in self._user_pseudonym_cache: digest = hashlib.sha256(f"{self.salt}:{raw_id}".encode("utf-8")).hexdigest() pseudonym = f"User_{digest[:6].upper()}" self._user_pseudonym_cache[raw_id] = pseudonym return self._user_pseudonym_cache[raw_id] def anonymize_text(self, text: str) -> Tuple[str, List[PIIEntity]]: """ Scans text, extracts PII entities, and applies redaction tokens. Returns: sanitized_text: String with all PII replaced by tokens. pii_entities: List of PIIEntity records. """ pii_list: List[PIIEntity] = [] sanitized = text # 1. Redact Emails for match in self.email_pattern.finditer(sanitized): pii_list.append(PIIEntity( entity_type="EMAIL", start_idx=match.start(), end_idx=match.end(), masked_value="[EMAIL_REDACTED]" )) sanitized = self.email_pattern.sub("[EMAIL_REDACTED]", sanitized) # 2. Redact Phone Numbers for match in self.phone_pattern.finditer(sanitized): pii_list.append(PIIEntity( entity_type="PHONE", start_idx=match.start(), end_idx=match.end(), masked_value="[PHONE_REDACTED]" )) sanitized = self.phone_pattern.sub("[PHONE_REDACTED]", sanitized) # 3. Redact National Identification Numbers (CC / DNI / SSN) for match in self.id_pattern.finditer(sanitized): pii_list.append(PIIEntity( entity_type="GOVERNMENT_ID", start_idx=match.start(), end_idx=match.end(), masked_value="[ID_REDACTED]" )) sanitized = self.id_pattern.sub("[ID_REDACTED]", sanitized) sanitized = self.ssn_pattern.sub("[ID_REDACTED]", sanitized) # 4. Redact Financial / Salary details for match in self.financial_pattern.finditer(sanitized): pii_list.append(PIIEntity( entity_type="FINANCIAL_SALARY", start_idx=match.start(), end_idx=match.end(), masked_value="[FINANCIAL_REDACTED]" )) sanitized = self.financial_pattern.sub("[FINANCIAL_REDACTED]", sanitized) # 5. Redact IP Addresses for match in self.ip_pattern.finditer(sanitized): pii_list.append(PIIEntity( entity_type="NETWORK_IP", start_idx=match.start(), end_idx=match.end(), masked_value="[IP_REDACTED]" )) sanitized = self.ip_pattern.sub("[IP_REDACTED]", sanitized) # 6. Redact Chat Mentions (@handle -> User_XXXX) def replace_mention(match): raw_handle = match.group(1) pseudo = self.get_pseudonym(raw_handle) pii_list.append(PIIEntity( entity_type="USER_MENTION", start_idx=match.start(), end_idx=match.end(), masked_value=pseudo )) return f"@{pseudo}" sanitized = self.mention_pattern.sub(replace_mention, sanitized) # 7. Redact Salutations with Personal Names ("Hola Carlos" -> "Hola [NAME_REDACTED]") def replace_greeting(match): greeting = match.group(1) name = match.group(2) # Avoid replacing known safe terms if name.lower() in ["equipo", "todos", "team", "all", "grupo"]: return match.group(0) pii_list.append(PIIEntity( entity_type="PERSON_NAME", start_idx=match.start(2), end_idx=match.end(2), masked_value="[NAME_REDACTED]" )) return f"{greeting} [NAME_REDACTED]" sanitized = self.greeting_name_pattern.sub(replace_greeting, sanitized) return sanitized, pii_list def anonymize_message(self, raw: RawMessage) -> SanitizedMessage: """Converts a RawMessage into a fully compliant SanitizedMessage.""" sender_pseudo = self.get_pseudonym(raw.sender_id) clean_text, detected_pii = self.anonymize_text(raw.text_content) return SanitizedMessage( message_id=raw.message_id, sender_pseudonym=sender_pseudo, department=raw.department, timestamp=raw.timestamp, sanitized_content=clean_text, channel=raw.channel, pii_removed_count=len(detected_pii), detected_pii=detected_pii ) def anonymize_batch(self, raw_messages: List[RawMessage]) -> List[SanitizedMessage]: """Processes a sequence of messages in memory with zero disk footprint.""" return [self.anonymize_message(m) for m in raw_messages]