Remove user agent checks

This commit is contained in:
2025-11-12 18:33:33 +03:00
Unverified
parent 019414608f
commit 68f3f805b2
5 changed files with 0 additions and 415 deletions
-66
View File
@@ -276,63 +276,6 @@ class AdminCLI:
table.add_row(entry) table.add_row(entry)
self.console.print(table) self.console.print(table)
def cmd_block_user_agent(self, args: List[str]) -> None:
if not args:
raise CLIError("Usage: block-user-agent <pattern> [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 <pattern> [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: def cmd_help(self) -> None:
cmds = { cmds = {
"login [username]": "Authenticate as owner/admin.", "login [username]": "Authenticate as owner/admin.",
@@ -344,9 +287,6 @@ class AdminCLI:
"block-word <words>": "Add words/phrases to chat filter.", "block-word <words>": "Add words/phrases to chat filter.",
"unblock-word <words>": "Remove words/phrases from filter.", "unblock-word <words>": "Remove words/phrases from filter.",
"blocklist": "Show current blocklist.", "blocklist": "Show current blocklist.",
"block-user-agent <patterns>": "Add user agent patterns to blocklist.",
"unblock-user-agent <patterns>": "Remove user agent patterns from blocklist.",
"user-agent-blocklist": "Show current user agent blocklist.",
"list": "List all users.", "list": "List all users.",
"user <user>": "Show detailed user information.", "user <user>": "Show detailed user information.",
"whoami": "Display current session context.", "whoami": "Display current session context.",
@@ -407,12 +347,6 @@ class AdminCLI:
self.cmd_unblock_word(args) self.cmd_unblock_word(args)
elif command == "blocklist": elif command == "blocklist":
self.cmd_list_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": elif command == "verify":
self.cmd_verify(args) self.cmd_verify(args)
elif command == "unverify": elif command == "unverify":
-29
View File
@@ -17,7 +17,6 @@ import os
from security.audit import log_security from security.audit import log_security
from security.profanity import contains_profanity 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 from security.rate_limit import rate_limit_per_ip
router = APIRouter() router = APIRouter()
@@ -73,20 +72,6 @@ def login(request: Request, login_request: LoginRequest, db: Session = Depends(g
client_ip = get_client_ip(request) client_ip = get_client_ip(request)
raw_ua = request.headers.get("user-agent") 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() user = db.query(User).filter(User.username == username).first()
if not user or not verify_password(login_request.password.strip(), user.password_hash): 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) client_ip = get_client_ip(request)
raw_ua = request.headers.get("user-agent") 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 # Determine if owner already exists
owner_exists = db.query(User).filter(User.username == OWNER_USERNAME).first() is not None owner_exists = db.query(User).filter(User.username == OWNER_USERNAME).first() is not None
-47
View File
@@ -7,13 +7,6 @@ from dependencies import get_current_user
from models import User from models import User
from security.audit import log_security from security.audit import log_security
from security.profanity import add_to_blocklist, get_blocklist, remove_from_blocklist 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): class BlocklistUpdateRequest(BaseModel):
@@ -66,44 +59,4 @@ def delete_from_blocklist(
return {"removed": removed, "words": updated} 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}
-28
View File
@@ -189,34 +189,6 @@ def _render_security(action: str, fields: Dict[str, Any]) -> List[str]:
total = len(fields.get("words") or []) total = len(fields.get("words") or [])
lines.append(f"Total entries: {total}") lines.append(f"Total entries: {total}")
return lines 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()}"] + [ return [f"{action.replace('_', ' ').capitalize()}"] + [
f"{key.replace('_', ' ').capitalize()}: {value}" f"{key.replace('_', ' ').capitalize()}: {value}"
for key, value in fields.items() for key, value in fields.items()
-245
View File
@@ -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