diff --git a/backend/routes/messaging.py b/backend/routes/messaging.py index 7f2a2cd..adc6348 100644 --- a/backend/routes/messaging.py +++ b/backend/routes/messaging.py @@ -26,7 +26,7 @@ import io import json from better_profanity import profanity as _bp from security.audit import log_access, log_dm, log_public_chat, log_security -from security.profanity import censor_text +from security.profanity import censor_text, contains_profanity from security.rate_limit import rate_limit_per_ip from websocket.utils import authenticate_user @@ -277,6 +277,9 @@ async def _send_message_internal( # Apply profanity filter before storing filtered_content = censor_text(raw_content) escaped_content = html.escape(filtered_content, quote=False) + + # Check if content was censored (use contains_profanity to detect actual profanity) + was_censored = contains_profanity(raw_content) if len(escaped_content) > 4096: raise HTTPException( @@ -368,17 +371,25 @@ async def _send_message_internal( _monitor_public_message_activity(current_user, filtered_content, db) message_payload = convert_message(new_message) - log_public_chat( - "message_created", - message_id=new_message.id, - user_id=current_user.id, - username=current_user.username, - reply_to=new_message.reply_to_id, - attachments=len(new_message.files or []), - length=len(new_message.content), - suspended=current_user.suspended, - content=new_message.content, - ) + + # Prepare log fields + log_fields = { + "message_id": new_message.id, + "user_id": current_user.id, + "username": current_user.username, + "reply_to": new_message.reply_to_id, + "attachments": len(new_message.files or []), + "length": len(new_message.content), + "suspended": current_user.suspended, + "content": new_message.content, + } + + # If content was censored, log both raw and censored versions + if was_censored: + log_fields["raw_content"] = raw_content + log_fields["censored_content"] = filtered_content + + log_public_chat("message_created", **log_fields) return {"status": "success", "message": message_payload} @@ -692,6 +703,10 @@ async def edit_message( original_content = message.content sanitized_content = censor_text(raw_content) escaped_content = html.escape(sanitized_content, quote=False) + + # Check if content was censored (use contains_profanity to detect actual profanity) + was_censored = contains_profanity(raw_content) + if len(escaped_content) > 4096: raise HTTPException(status_code=400, detail="Message too long") @@ -702,15 +717,23 @@ async def edit_message( db.refresh(message) payload = convert_message(message) - log_public_chat( - "message_edited", - message_id=message.id, - user_id=current_user.id, - username=current_user.username, - reply_to=message.reply_to_id, - content=message.content, - previous_content=original_content, - ) + + # Prepare log fields + log_fields = { + "message_id": message.id, + "user_id": current_user.id, + "username": current_user.username, + "reply_to": message.reply_to_id, + "content": message.content, + "previous_content": original_content, + } + + # If content was censored, log both raw and censored versions + if was_censored: + log_fields["raw_content"] = raw_content + log_fields["censored_content"] = sanitized_content + + log_public_chat("message_edited", **log_fields) return {"status": "success", "message": payload} diff --git a/backend/security/audit.py b/backend/security/audit.py index 52bad60..f9e7e64 100644 --- a/backend/security/audit.py +++ b/backend/security/audit.py @@ -204,7 +204,16 @@ def _render_public_chat(action: str, fields: Dict[str, Any]) -> List[str]: attachments = fields.get("attachments") if attachments: lines.append(f"Attachments: {_plural('file', attachments)}") - if fields.get("content"): + + # If content was censored, log both raw and censored versions + if fields.get("raw_content") is not None: + lines.append("Raw content (before censoring):") + for line in unescape(fields["raw_content"]).splitlines(): + lines.append(f"| {line}") + lines.append("Censored content (stored):") + for line in unescape(fields.get("censored_content", fields.get("content", ""))).splitlines(): + lines.append(f"| {line}") + elif fields.get("content"): lines.append("Content:") for line in unescape(fields["content"]).splitlines(): lines.append(f"| {line}") @@ -217,7 +226,16 @@ def _render_public_chat(action: str, fields: Dict[str, Any]) -> List[str]: lines.append("Previous content:") for line in unescape(fields["previous_content"] or "").splitlines() or [""]: lines.append(f"| {line}") - if fields.get("content"): + + # If content was censored, log both raw and censored versions + if fields.get("raw_content") is not None: + lines.append("Raw content (before censoring):") + for line in unescape(fields["raw_content"]).splitlines(): + lines.append(f"| {line}") + lines.append("Censored content (stored):") + for line in unescape(fields.get("censored_content", fields.get("content", ""))).splitlines(): + lines.append(f"| {line}") + elif fields.get("content"): lines.append("New content:") for line in unescape(fields["content"] or "").splitlines() or [""]: lines.append(f"| {line}") diff --git a/backend/security/profanity.py b/backend/security/profanity.py index d7adffb..b2f3acb 100644 --- a/backend/security/profanity.py +++ b/backend/security/profanity.py @@ -713,54 +713,36 @@ def censor_text(text: str) -> str: def contains_profanity(text: str) -> bool: + """ + Check if text contains profanity that would be censored. + Returns True if censor_text would actually censor anything. + """ if not text: return False - _rebuild_dictionary() + # Use censor_text to check if anything would be censored + # This ensures consistency between contains_profanity and censor_text + censored = censor_text(text) - # Extract only alphanumeric characters and normalize homoglyphs - # This removes special characters, emojis, etc. that could be used to bypass the filter - normalized_text, _ = _extract_alphanumeric_with_mapping(text) - normalized_lower = normalized_text.lower() + # Check if any characters were actually censored (changed to asterisks) + # by comparing the original text with the censored version + # We need to account for the fact that the original might already contain asterisks + if censored == text: + return False # No changes, so no profanity - # Check phrase patterns on normalized text (to handle special characters) - for pattern in _PHRASE_PATTERNS: - if pattern.search(normalized_lower): - return True - if _find_fuzzy_phrase_spans(normalized_lower, "generic"): - return True + # If the text changed, check if any non-asterisk characters were replaced + # by comparing character-by-character (excluding positions that were already asterisks) + for i, (orig_char, censored_char) in enumerate(zip(text, censored)): + if orig_char != "*" and censored_char == "*": + return True # A non-asterisk character was censored - # Check for profane words as substrings/subsequences (to catch cases like "хуй" in "хууй" or "хуйня") - profane_words = _STATIC_TERMS - substring_spans = _check_profanity_substrings(normalized_text, profane_words) - - if substring_spans: - # Check if any found profanity is not part of a whitelisted word - for span_start, span_end in substring_spans: - is_whitelisted = False - for whitelist_word in _WHITELIST: - normalized_whitelist, _ = _extract_alphanumeric_with_mapping(whitelist_word) - normalized_whitelist_lower = normalized_whitelist.lower() - wl_pos = normalized_lower.find(normalized_whitelist_lower) - if wl_pos != -1: - # Check if profane span is within whitelisted word - if wl_pos <= span_start < wl_pos + len(normalized_whitelist_lower): - is_whitelisted = True - break - if not is_whitelisted: + # If censored is longer, check the extra characters + if len(censored) > len(text): + for i in range(len(text), len(censored)): + if censored[i] == "*": return True - # Remove whitelisted words from text before checking profanity - # This allows standalone whitelisted words but still blocks them in phrases - for whitelist_word in _WHITELIST: - # Normalize whitelist word too - normalized_whitelist, _ = _extract_alphanumeric_with_mapping(whitelist_word) - normalized_whitelist_lower = normalized_whitelist.lower() - # Use word boundaries to match whole words only - pattern = re.compile(r"\b" + re.escape(normalized_whitelist_lower) + r"\b", re.IGNORECASE) - normalized_lower = pattern.sub("", normalized_lower) - - return _profanity.contains_profanity(normalized_lower) + return False def contains_sensitive_phrase(text: str) -> bool: