diff --git a/backend/admin_cli.py b/backend/admin_cli.py index 0f612e2..6afd4f8 100644 --- a/backend/admin_cli.py +++ b/backend/admin_cli.py @@ -276,63 +276,6 @@ class AdminCLI: table.add_row(entry) self.console.print(table) - def cmd_block_user_agent(self, args: List[str]) -> None: - if not args: - raise CLIError("Usage: block-user-agent [additional patterns...]") - self._require_auth() - patterns = args - response = self._request("POST", "moderation/user-agent-blocklist", json={"words": patterns}) - data = response.json() - added = data.get("added", []) - current = data.get("patterns", []) - if added: - self.console.print(f"[bold green]Added {len(added)} pattern{'s' if len(added) != 1 else ''} to user agent blocklist.[/]") - else: - self.console.print("[yellow]No new patterns added.[/]") - self.console.print(f"Blocklist size: {len(current)}") - - def cmd_unblock_user_agent(self, args: List[str]) -> None: - if not args: - raise CLIError("Usage: unblock-user-agent [additional patterns...]") - self._require_auth() - response = self._request("DELETE", "moderation/user-agent-blocklist", json={"words": args}) - data = response.json() - removed = data.get("removed", []) - current = data.get("patterns", []) - if removed: - self.console.print(f"[bold green]Removed {len(removed)} pattern{'s' if len(removed) != 1 else ''} from user agent blocklist.[/]") - else: - self.console.print("[yellow]No matching patterns removed.[/]") - self.console.print(f"Blocklist size: {len(current)}") - - def cmd_list_user_agent_blocklist(self) -> None: - self._require_auth() - response = self._request("GET", "moderation/user-agent-blocklist") - data = response.json() - static = data.get("static", []) - external = data.get("external", []) - - if not static and not external: - self.console.print("[cyan]User agent blocklist is empty.[/]") - return - - if static: - table_static = Table(title="Static Blocked User Agent Patterns", show_lines=True) - table_static.add_column("Pattern", style="yellow") - for entry in static: - table_static.add_row(entry) - self.console.print(table_static) - - if external: - table_external = Table(title="External Blocked User Agent Patterns", show_lines=True) - table_external.add_column("Pattern", style="cyan") - for entry in external: - table_external.add_row(entry) - self.console.print(table_external) - - if not external: - self.console.print("[dim]No external patterns. Use 'block-user-agent' to add patterns.[/]") - def cmd_help(self) -> None: cmds = { "login [username]": "Authenticate as owner/admin.", @@ -344,9 +287,6 @@ class AdminCLI: "block-word ": "Add words/phrases to chat filter.", "unblock-word ": "Remove words/phrases from filter.", "blocklist": "Show current blocklist.", - "block-user-agent ": "Add user agent patterns to blocklist.", - "unblock-user-agent ": "Remove user agent patterns from blocklist.", - "user-agent-blocklist": "Show current user agent blocklist.", "list": "List all users.", "user ": "Show detailed user information.", "whoami": "Display current session context.", @@ -407,12 +347,6 @@ class AdminCLI: self.cmd_unblock_word(args) elif command == "blocklist": self.cmd_list_blocklist() - elif command == "block-user-agent": - self.cmd_block_user_agent(args) - elif command == "unblock-user-agent": - self.cmd_unblock_user_agent(args) - elif command == "user-agent-blocklist": - self.cmd_list_user_agent_blocklist() elif command == "verify": self.cmd_verify(args) elif command == "unverify": diff --git a/backend/routes/account.py b/backend/routes/account.py index 8042c20..02fd8d3 100644 --- a/backend/routes/account.py +++ b/backend/routes/account.py @@ -17,7 +17,6 @@ import os from security.audit import log_security from security.profanity import contains_profanity -from security.user_agent_blocklist import is_user_agent_blocked from security.rate_limit import rate_limit_per_ip router = APIRouter() @@ -73,20 +72,6 @@ def login(request: Request, login_request: LoginRequest, db: Session = Depends(g client_ip = get_client_ip(request) raw_ua = request.headers.get("user-agent") - if is_user_agent_blocked(raw_ua): - log_security( - "blocked_user_agent", - severity="warning", - username=username, - ip=client_ip, - user_agent=raw_ua or "Unknown", - action_type="login", - ) - raise HTTPException( - status_code=403, - detail="Доступ запрещён" - ) - user = db.query(User).filter(User.username == username).first() if not user or not verify_password(login_request.password.strip(), user.password_hash): @@ -189,20 +174,6 @@ def register(request: Request, register_request: RegisterRequest, db: Session = client_ip = get_client_ip(request) raw_ua = request.headers.get("user-agent") - if is_user_agent_blocked(raw_ua): - log_security( - "blocked_user_agent", - severity="warning", - username=username, - ip=client_ip, - user_agent=raw_ua or "Unknown", - action_type="registration", - ) - raise HTTPException( - status_code=403, - detail="Доступ запрещён" - ) - # Determine if owner already exists owner_exists = db.query(User).filter(User.username == OWNER_USERNAME).first() is not None diff --git a/backend/routes/moderation.py b/backend/routes/moderation.py index 6b07fea..7eee746 100644 --- a/backend/routes/moderation.py +++ b/backend/routes/moderation.py @@ -7,13 +7,6 @@ from dependencies import get_current_user from models import User from security.audit import log_security from security.profanity import add_to_blocklist, get_blocklist, remove_from_blocklist -from security.user_agent_blocklist import ( - add_to_blocklist as add_ua_to_blocklist, - get_blocklist as get_ua_blocklist, - get_static_blocklist as get_ua_static_blocklist, - get_external_blocklist as get_ua_external_blocklist, - remove_from_blocklist as remove_ua_from_blocklist, -) class BlocklistUpdateRequest(BaseModel): @@ -66,44 +59,4 @@ def delete_from_blocklist( return {"removed": removed, "words": updated} -@router.get("/user-agent-blocklist") -def list_user_agent_blocklist(current_user: User = Depends(get_current_user)): - _ensure_owner(current_user) - return { - "patterns": get_ua_blocklist(), - "static": get_ua_static_blocklist(), - "external": get_ua_external_blocklist(), - } - - -@router.post("/user-agent-blocklist") -def append_user_agent_blocklist( - request: BlocklistUpdateRequest, - current_user: User = Depends(get_current_user) -): - _ensure_owner(current_user) - added, updated = add_ua_to_blocklist(request.words) - log_security( - "user_agent_blocklist_add", - actor=current_user.username, - actor_id=current_user.id, - added=added, - ) - return {"added": added, "patterns": updated} - - -@router.delete("/user-agent-blocklist") -def delete_from_user_agent_blocklist( - request: BlocklistUpdateRequest, - current_user: User = Depends(get_current_user) -): - _ensure_owner(current_user) - removed, updated = remove_ua_from_blocklist(request.words) - log_security( - "user_agent_blocklist_remove", - actor=current_user.username, - actor_id=current_user.id, - removed=removed, - ) - return {"removed": removed, "patterns": updated} diff --git a/backend/security/audit.py b/backend/security/audit.py index cf30fdc..52bad60 100644 --- a/backend/security/audit.py +++ b/backend/security/audit.py @@ -189,34 +189,6 @@ def _render_security(action: str, fields: Dict[str, Any]) -> List[str]: total = len(fields.get("words") or []) lines.append(f"Total entries: {total}") return lines - if action == "blocked_user_agent": - action_type = fields.get("action_type", "access") - lines = [f"Blocked user agent attempted {action_type}"] - if fields.get("username"): - lines.append(f"Username: {fields['username']}") - if fields.get("user_agent"): - lines.append(f"User agent: {fields['user_agent']}") - if fields.get("ip"): - ip_raw = fields["ip"] - ip_display = "localhost" if ip_raw in {"127.0.0.1", "::1"} else ip_raw - lines.append(f"IP: {ip_display}") - return lines - if action == "user_agent_blocklist_add": - added = fields.get("added") or [] - lines = [f"User agent blocklist updated by {_format_actor(fields, 'actor')}"] - if added: - lines.append(f"Added patterns: {', '.join(added)}") - total = len(fields.get("patterns") or []) - lines.append(f"Total patterns: {total}") - return lines - if action == "user_agent_blocklist_remove": - removed = fields.get("removed") or [] - lines = [f"User agent blocklist cleaned by {_format_actor(fields, 'actor')}"] - if removed: - lines.append(f"Removed patterns: {', '.join(removed)}") - total = len(fields.get("patterns") or []) - lines.append(f"Total patterns: {total}") - return lines return [f"{action.replace('_', ' ').capitalize()}"] + [ f"{key.replace('_', ' ').capitalize()}: {value}" for key, value in fields.items() diff --git a/backend/security/user_agent_blocklist.py b/backend/security/user_agent_blocklist.py deleted file mode 100644 index 692c5e8..0000000 --- a/backend/security/user_agent_blocklist.py +++ /dev/null @@ -1,245 +0,0 @@ -from __future__ import annotations - -import json -import re -from pathlib import Path -from threading import RLock -from typing import Iterable, List, Set -from user_agents import parse as parse_ua - -BLOCKLIST_PATH = Path("data/user_agent_blocklist.json") -BLOCKLIST_PATH.parent.mkdir(parents=True, exist_ok=True) - -# Hardcoded list of known bot/scraper user agents -_STATIC_BLOCKED_AGENTS: Set[str] = { - "python-requests", - "python requests", - "requests", - "curl", - "wget", - "httpie", - "go-http-client", - "java/", - "okhttp", - "apache-httpclient", - "scrapy", - "mechanize", - "beautifulsoup", - "urllib", - "httpx", - "aiohttp", - "postman", - "insomnia", - "postmanruntime", - "restclient", - "http", - "bot", - "crawler", - "spider", - "scraper", -} - -_blocklist_lock = RLock() -_blocklist_cache: Set[str] | None = None - - -def _normalize_pattern(pattern: str) -> str: - cleaned = re.sub(r"\s+", " ", str(pattern)).strip().lower() - return cleaned - - -def _load_blocklist() -> Set[str]: - global _blocklist_cache - with _blocklist_lock: - if _blocklist_cache is not None: - return _blocklist_cache - - # Start with static hardcoded patterns - patterns = set(_normalize_pattern(p) for p in _STATIC_BLOCKED_AGENTS) - - # Load additional patterns from external file - if BLOCKLIST_PATH.exists(): - try: - data = json.loads(BLOCKLIST_PATH.read_text(encoding="utf-8")) - if isinstance(data, list): - external_patterns = set(_normalize_pattern(p) for p in data if p) - patterns.update(external_patterns) - except Exception: - pass - - _blocklist_cache = patterns - return patterns - - -def _write_blocklist(external_patterns: Iterable[str]) -> None: - """Write only external patterns to the JSON file. Static patterns are not stored.""" - normalized = sorted(set(_normalize_pattern(p) for p in external_patterns if p)) - BLOCKLIST_PATH.write_text( - json.dumps(normalized, ensure_ascii=False, indent=2) + "\n", - encoding="utf-8" - ) - # Clear cache so it reloads with static + external patterns - global _blocklist_cache - with _blocklist_lock: - _blocklist_cache = None - - -def _match_pattern(text: str, pattern: str) -> bool: - normalized_text = text.lower() - normalized_pattern = pattern.lower() - - if normalized_pattern in normalized_text: - return True - - try: - regex = re.compile(normalized_pattern, re.IGNORECASE) - if regex.search(normalized_text): - return True - except re.error: - pass - - return False - - -def is_user_agent_blocked(raw_user_agent: str | None) -> bool: - if not raw_user_agent: - return False - - blocklist = _load_blocklist() - if not blocklist: - return False - - for pattern in blocklist: - if _match_pattern(raw_user_agent, pattern): - return True - - try: - ua = parse_ua(raw_user_agent) - browser_name = ua.browser.family or "" - os_name = ua.os.family or "" - - browser_pattern = browser_name.lower() if browser_name else "" - os_pattern = os_name.lower() if os_name else "" - - formatted = f"{os_name or 'Other'}, {browser_name or 'Unknown browser'}" - if ua.browser.version_string: - formatted = f"{formatted} {ua.browser.version_string}" - - for pattern in blocklist: - if _match_pattern(formatted, pattern): - return True - if browser_pattern and _match_pattern(browser_pattern, pattern): - return True - if os_pattern and _match_pattern(os_pattern, pattern): - return True - except Exception: - pass - - return False - - -def get_blocklist() -> List[str]: - """Get all blocked patterns (static + external).""" - with _blocklist_lock: - return sorted(_load_blocklist()) - - -def get_static_blocklist() -> List[str]: - """Get only the hardcoded static patterns.""" - return sorted(_normalize_pattern(p) for p in _STATIC_BLOCKED_AGENTS) - - -def get_external_blocklist() -> List[str]: - """Get only the patterns from the external JSON file.""" - if not BLOCKLIST_PATH.exists(): - return [] - try: - data = json.loads(BLOCKLIST_PATH.read_text(encoding="utf-8")) - if isinstance(data, list): - return sorted(_normalize_pattern(p) for p in data if p) - except Exception: - pass - return [] - - -def add_to_blocklist(patterns: Iterable[str]) -> tuple[List[str], List[str]]: - """Add patterns to the external blocklist. Static patterns cannot be modified.""" - normalized = set(_normalize_pattern(p) for p in patterns if p) - if not normalized: - return [], get_blocklist() - - with _blocklist_lock: - # Only add to external blocklist, not static - static_patterns = set(_normalize_pattern(p) for p in _STATIC_BLOCKED_AGENTS) - - # Filter out static patterns (they're already blocked) - normalized = normalized - static_patterns - if not normalized: - return [], get_blocklist() - - # Load current external patterns - external_current = set() - if BLOCKLIST_PATH.exists(): - try: - data = json.loads(BLOCKLIST_PATH.read_text(encoding="utf-8")) - if isinstance(data, list): - external_current = set(_normalize_pattern(p) for p in data if p) - except Exception: - pass - - added = sorted(normalized - external_current) - if not added: - return [], get_blocklist() - - updated_external = sorted(external_current | normalized) - _write_blocklist(updated_external) - - # Clear cache to reload - _blocklist_cache = None - - return added, get_blocklist() - - -def remove_from_blocklist(patterns: Iterable[str]) -> tuple[List[str], List[str]]: - """Remove patterns from the external blocklist. Static patterns cannot be removed.""" - normalized = set(_normalize_pattern(p) for p in patterns if p) - if not normalized: - return [], get_blocklist() - - with _blocklist_lock: - # Only remove from external blocklist, not static - static_patterns = set(_normalize_pattern(p) for p in _STATIC_BLOCKED_AGENTS) - - # Filter out static patterns (cannot remove them) - normalized = normalized - static_patterns - if not normalized: - return [], get_blocklist() - - # Load current external patterns - external_current = set() - if BLOCKLIST_PATH.exists(): - try: - data = json.loads(BLOCKLIST_PATH.read_text(encoding="utf-8")) - if isinstance(data, list): - external_current = set(_normalize_pattern(p) for p in data if p) - except Exception: - pass - - removed = sorted(pattern for pattern in normalized if pattern in external_current) - if not removed: - return [], get_blocklist() - - updated_external = sorted(external_current - normalized) - _write_blocklist(updated_external) - - # Clear cache to reload - _blocklist_cache = None - - return removed, get_blocklist() - - -def clear_blocklist_cache() -> None: - global _blocklist_cache - with _blocklist_lock: - _blocklist_cache = None -