Compare commits

...
Sign in to create a new pull request.

30 commits

Author SHA1 Message Date
Gros Frumos
cb89a90771 fix: viewport safe-area-inset for iOS PWA + disable pinch zoom
Topbar (avatar, network indicator) was hidden behind iOS status bar
in standalone PWA mode. Added safe-area-inset-top padding to topbar.
Disabled user-scalable to prevent accidental zoom.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-21 16:40:30 +02:00
Gros Frumos
6e2503dc3f fix: lock viewport height to prevent topbar scroll-off
Body fixed to 100dvh with overflow:hidden. SOS button capped at
min(60vmin, 70vw, 300px) to fit within viewport alongside topbar.
Bump SW cache to v3.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-21 16:25:21 +02:00
Gros Frumos
5da2a9a708 infra: add Docker setup for portable deployment
Dockerfile (Python 3.12 slim) + docker-compose (backend + nginx).
Backend on port 8000 inside container, nginx proxies API and serves
frontend static. SQLite persisted in named volume. Nginx listens on
127.0.0.1:8080 — external SSL handled by host reverse proxy.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-21 16:23:08 +02:00
Gros Frumos
6617c85cd5 fix: bump SW cache version to force app.js refresh
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-21 16:13:08 +02:00
Gros Frumos
268fb62bf3 feat: test signal via avatar/indicator tap on main screen
Tapping user avatar or network indicator sends a test signal with
geo data. Backend formats it as "Тест от username" (🧪) instead of
"Сигнал" (🚨). Only active after login on main screen.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-21 16:06:02 +02:00
Gros Frumos
0562cb4e47 sec: server-side email domain check + IP block on violations
Only @tutlot.com emails allowed for registration (checked server-side,
invisible to frontend inspect). Wrong domain → scary message + IP
violation tracked. 5 violations → IP permanently blocked from login
and registration. Block screen with OK button on frontend.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-21 15:58:16 +02:00
Gros Frumos
47b89ded8d feat: geo location as Google Maps link in Telegram notifications
When signal has geo, show clickable Google Maps link instead of raw
coordinates. Without geo, show "Гео нету". Added parse_mode=HTML
to send_message for link rendering.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-21 14:21:41 +02:00
Gros Frumos
04f7bd79e2 auth: replace UUID-based login with JWT credential verification
Login now requires login/email + password verified against DB via
/api/auth/login. Only approved registrations can access the app.
Signal endpoint accepts JWT Bearer tokens alongside legacy api_key auth.
Old UUID-only registration flow removed from frontend.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-21 14:14:12 +02:00
Gros Frumos
1adcabf3a6 kin: BATON-008 На главной странице под логином сделать кнопку модулем регистрации - указать почту, логин и пароль, нажать зарегистрироваться. После этого сообщение о регистрации приходит в чат администратору 5694335584 и кнопка апрув или не апрув, если апрув то отправителя улетает пуш на pwa что он зарегистрирован, если отказ то ничего не происходит 2026-03-21 13:49:57 +02:00
Gros Frumos
baf05b6d84 kin: BATON-BIZ-004 Удалить дублирующую настройку логирования в telegram.py 2026-03-21 13:49:57 +02:00
Gros Frumos
6444b30d17 kin: BATON-BIZ-002 Убрать hardcoded VAPID key из meta-тега, читать с /api/push/public-key 2026-03-21 13:49:57 +02:00
Gros Frumos
ea06309a6e kin: BATON-BIZ-001-backend_dev 2026-03-21 13:49:57 +02:00
Gros Frumos
e266b6506e kin: BATON-BIZ-004-backend_dev 2026-03-21 13:49:57 +02:00
Gros Frumos
86a41a3b35 kin: BATON-BIZ-002-frontend_dev 2026-03-21 13:49:57 +02:00
Gros Frumos
40e1a9fa48 kin: BATON-008 На главной странице под логином сделать кнопку модулем регистрации - указать почту, логин и пароль, нажать зарегистрироваться. После этого сообщение о регистрации приходит в чат администратору 5694335584 и кнопка апрув или не апрув, если апрув то отправителя улетает пуш на pwa что он зарегистрирован, если отказ то ничего не происходит 2026-03-21 13:49:57 +02:00
Gros Frumos
8c4c46ee92 kin: BATON-FIX-016 [TECH DEBT] VAPID public key жёстко вшит как пустая строка в <meta>-тег — требует ручного заполнения при деплое 2026-03-21 13:49:57 +02:00
Gros Frumos
5fe9a603f8 kin: BATON-FIX-016-frontend_dev 2026-03-21 13:49:57 +02:00
Kin Agent
5fa3a35d27 fix: add ExecStartPre pip install to baton.service — prevents manual package installs
Fixes BATON-FIX-015: email-validator was installed manually as root because
deploy process had no pip install step. Added ExecStartPre to run
pip install -r requirements.txt on every service start/restart.
2026-03-21 09:17:06 +00:00
Gros Frumos
debd7895f4 kin: BATON-SEC-001 httpcore suppress in main.py 2026-03-21 10:56:55 +02:00
Gros Frumos
635991078c sec: suppress httpcore transport logger in main.py
Дублирует аналогичный fix в telegram.py — httpcore тоже логирует
URLs с BOT_TOKEN на transport уровне. Синхронизировано с ручным
патчем на сервере.

Refs: #1303, #1309

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-21 10:56:52 +02:00
Gros Frumos
a0dc6a7b22 kin: BATON-SEC-001 pre-commit hook + httpx logging hardening 2026-03-21 10:56:01 +02:00
Gros Frumos
dd556e2f05 sec: pre-commit hook + httpx exception logging hardening
1. .pre-commit-config.yaml — local pygrep hook блокирует коммиты
   с токенами формата \d{9,10}:AA[A-Za-z0-9_-]{35} (Telegram bot tokens).
   Проверено: срабатывает на токен, пропускает чистые файлы.

2. backend/telegram.py — три функции (send_registration_notification,
   answer_callback_query, edit_message_text) логировали exc напрямую,
   что раскрывало BOT_TOKEN в URL httpx-исключений в journalctl.
   Заменено на type(exc).__name__ — только тип ошибки, без URL.

Refs: #1303, #1309, #1283

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-21 10:55:34 +02:00
Gros Frumos
5401363ea9 kin: BATON-FIX-013 CORS allow_methods: добавить GET для /health эндпоинтов 2026-03-21 09:37:57 +02:00
Gros Frumos
c7661d7c1e Merge branch 'BATON-008-backend_dev' 2026-03-21 09:34:21 +02:00
Gros Frumos
fde7f57a7a kin: BATON-008-backend_dev 2026-03-21 09:34:21 +02:00
Gros Frumos
35eef641fd Merge branch 'BATON-FIX-013-backend_dev' 2026-03-21 09:34:18 +02:00
Gros Frumos
283ff61dc5 fix: sync allow_methods с main — добавить HEAD и OPTIONS 2026-03-21 09:33:53 +02:00
Gros Frumos
6d5d84a882 fix: CORS allow_methods добавить GET для /health эндпоинтов
CORSMiddleware: allow_methods=['POST'] → ['GET', 'POST']
Позволяет браузерам делать GET-запросы к /health и /api/health без CORS-блокировки.

BATON-FIX-013
2026-03-21 09:33:09 +02:00
Gros Frumos
257631436a Merge branch 'BATON-FIX-007-backend_dev' 2026-03-21 09:30:44 +02:00
Gros Frumos
b2fecc5993 kin: BATON-FIX-007-backend_dev 2026-03-21 09:30:44 +02:00
33 changed files with 2836 additions and 432 deletions

16
.dockerignore Normal file
View file

@ -0,0 +1,16 @@
.git
.gitignore
.env
.venv
venv
__pycache__
*.pyc
*.db
tests/
docs/
deploy/
frontend/
nginx/
*.md
.kin_worktrees/
PROGRESS.md

View file

@ -11,3 +11,8 @@ DB_PATH=baton.db
# CORS
FRONTEND_ORIGIN=https://yourdomain.com
# VAPID Push Notifications (generate with: python -c "from py_vapid import Vapid; v=Vapid(); v.generate_keys(); print(v.public_key, v.private_key)")
VAPID_PUBLIC_KEY=
VAPID_PRIVATE_KEY=
VAPID_CLAIMS_EMAIL=

11
.pre-commit-config.yaml Normal file
View file

@ -0,0 +1,11 @@
repos:
- repo: local
hooks:
- id: no-telegram-bot-token
name: Block Telegram bot tokens
# Matches tokens of format: 1234567890:AAFisjLS-yO_AmwqMjpBQgfV9qlHnexZlMs
# Pattern: 9-10 digits, colon, "AA", then 35 alphanumeric/dash/underscore chars
entry: '\d{9,10}:AA[A-Za-z0-9_-]{35}'
language: pygrep
types: [text]
exclude: '^\.pre-commit-config\.yaml$'

12
Dockerfile Normal file
View file

@ -0,0 +1,12 @@
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY backend/ backend/
EXPOSE 8000
CMD ["uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", "8000"]

View file

@ -22,7 +22,9 @@ WEBHOOK_ENABLED: bool = os.getenv("WEBHOOK_ENABLED", "true").lower() == "true"
FRONTEND_ORIGIN: str = os.getenv("FRONTEND_ORIGIN", "http://localhost:3000")
APP_URL: str | None = os.getenv("APP_URL") # Публичный URL приложения для keep-alive self-ping
ADMIN_TOKEN: str = _require("ADMIN_TOKEN")
ADMIN_CHAT_ID: str = os.getenv("ADMIN_CHAT_ID", "5694335584")
ADMIN_CHAT_ID: str = _require("ADMIN_CHAT_ID")
VAPID_PRIVATE_KEY: str = os.getenv("VAPID_PRIVATE_KEY", "")
VAPID_PUBLIC_KEY: str = os.getenv("VAPID_PUBLIC_KEY", "")
VAPID_CLAIMS_EMAIL: str = os.getenv("VAPID_CLAIMS_EMAIL", "")
JWT_SECRET: str = os.getenv("JWT_SECRET", "")
JWT_TOKEN_EXPIRE_SECONDS: int = int(os.getenv("JWT_TOKEN_EXPIRE_SECONDS", "2592000")) # 30 days

View file

@ -84,6 +84,14 @@ async def init_db() -> None:
ON registrations(email);
CREATE INDEX IF NOT EXISTS idx_registrations_login
ON registrations(login);
CREATE TABLE IF NOT EXISTS ip_blocks (
ip TEXT NOT NULL PRIMARY KEY,
violation_count INTEGER NOT NULL DEFAULT 0,
is_blocked INTEGER NOT NULL DEFAULT 0,
created_at TEXT DEFAULT (datetime('now')),
blocked_at TEXT DEFAULT NULL
);
""")
# Migrations for existing databases (silently ignore if columns already exist)
for stmt in [
@ -341,9 +349,10 @@ async def get_registration(reg_id: int) -> Optional[dict]:
async def update_registration_status(reg_id: int, status: str) -> bool:
"""Update registration status only if currently 'pending'. Returns False if already processed."""
async with _get_conn() as conn:
async with conn.execute(
"UPDATE registrations SET status = ? WHERE id = ?",
"UPDATE registrations SET status = ? WHERE id = ? AND status = 'pending'",
(status, reg_id),
) as cur:
changed = cur.rowcount > 0
@ -351,6 +360,30 @@ async def update_registration_status(reg_id: int, status: str) -> bool:
return changed
async def get_registration_by_login_or_email(login_or_email: str) -> Optional[dict]:
async with _get_conn() as conn:
async with conn.execute(
"""
SELECT id, email, login, password_hash, status, push_subscription, created_at
FROM registrations
WHERE login = ? OR email = ?
""",
(login_or_email, login_or_email),
) as cur:
row = await cur.fetchone()
if row is None:
return None
return {
"id": row["id"],
"email": row["email"],
"login": row["login"],
"password_hash": row["password_hash"],
"status": row["status"],
"push_subscription": row["push_subscription"],
"created_at": row["created_at"],
}
async def save_telegram_batch(
message_text: str,
signals_count: int,
@ -373,3 +406,36 @@ async def save_telegram_batch(
)
await conn.commit()
return batch_id
async def is_ip_blocked(ip: str) -> bool:
async with _get_conn() as conn:
async with conn.execute(
"SELECT is_blocked FROM ip_blocks WHERE ip = ?", (ip,)
) as cur:
row = await cur.fetchone()
return bool(row["is_blocked"]) if row else False
async def record_ip_violation(ip: str) -> int:
"""Increment violation count for IP. Returns new count. Blocks IP at threshold."""
async with _get_conn() as conn:
await conn.execute(
"""
INSERT INTO ip_blocks (ip, violation_count) VALUES (?, 1)
ON CONFLICT(ip) DO UPDATE SET violation_count = violation_count + 1
""",
(ip,),
)
async with conn.execute(
"SELECT violation_count FROM ip_blocks WHERE ip = ?", (ip,)
) as cur:
row = await cur.fetchone()
count = row["violation_count"]
if count >= 5:
await conn.execute(
"UPDATE ip_blocks SET is_blocked = 1, blocked_at = datetime('now') WHERE ip = ?",
(ip,),
)
await conn.commit()
return count

View file

@ -2,6 +2,7 @@ from __future__ import annotations
import asyncio
import hashlib
import hmac
import logging
import os
import secrets
@ -16,11 +17,24 @@ from fastapi.responses import JSONResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from backend import config, db, push, telegram
from backend.middleware import rate_limit_auth_register, rate_limit_register, rate_limit_signal, verify_admin_token, verify_webhook_secret
from backend.middleware import (
_get_client_ip,
_verify_jwt_token,
check_ip_not_blocked,
create_auth_token,
rate_limit_auth_login,
rate_limit_auth_register,
rate_limit_register,
rate_limit_signal,
verify_admin_token,
verify_webhook_secret,
)
from backend.models import (
AdminBlockRequest,
AdminCreateUserRequest,
AdminSetPasswordRequest,
AuthLoginRequest,
AuthLoginResponse,
AuthRegisterRequest,
AuthRegisterResponse,
RegisterRequest,
@ -33,6 +47,7 @@ _api_key_bearer = HTTPBearer(auto_error=False)
logging.basicConfig(level=logging.INFO)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
@ -50,6 +65,18 @@ def _hash_password(password: str) -> str:
dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 260_000)
return f"{salt.hex()}:{dk.hex()}"
def _verify_password(password: str, stored_hash: str) -> bool:
"""Verify a password against a stored PBKDF2-HMAC-SHA256 hash (salt_hex:dk_hex)."""
try:
salt_hex, dk_hex = stored_hash.split(":", 1)
salt = bytes.fromhex(salt_hex)
expected_dk = bytes.fromhex(dk_hex)
actual_dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 260_000)
return hmac.compare_digest(actual_dk, expected_dk)
except Exception:
return False
# aggregator = telegram.SignalAggregator(interval=10) # v2.0 feature — отключено в v1 (ADR-004)
_KEEPALIVE_INTERVAL = 600 # 10 минут
@ -120,7 +147,7 @@ app = FastAPI(lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=[config.FRONTEND_ORIGIN],
allow_methods=["POST"],
allow_methods=["GET", "HEAD", "OPTIONS", "POST"],
allow_headers=["Content-Type", "Authorization"],
)
@ -131,6 +158,12 @@ async def health() -> dict[str, Any]:
return {"status": "ok"}
@app.get("/api/vapid-public-key")
@app.get("/api/push/public-key")
async def vapid_public_key() -> dict[str, str]:
return {"vapid_public_key": config.VAPID_PUBLIC_KEY}
@app.post("/api/register", response_model=RegisterResponse)
async def register(body: RegisterRequest, _: None = Depends(rate_limit_register)) -> RegisterResponse:
api_key = secrets.token_hex(32)
@ -146,13 +179,36 @@ async def signal(
) -> SignalResponse:
if credentials is None:
raise HTTPException(status_code=401, detail="Unauthorized")
key_hash = _hash_api_key(credentials.credentials)
stored_hash = await db.get_api_key_hash_by_uuid(body.user_id)
if stored_hash is None or not secrets.compare_digest(key_hash, stored_hash):
raise HTTPException(status_code=401, detail="Unauthorized")
if await db.is_user_blocked(body.user_id):
raise HTTPException(status_code=403, detail="User is blocked")
user_identifier: str = ""
user_name: str = ""
# Try JWT auth first (new registration flow)
jwt_payload = None
try:
jwt_payload = _verify_jwt_token(credentials.credentials)
except Exception:
pass
if jwt_payload is not None:
reg_id = int(jwt_payload["sub"])
reg = await db.get_registration(reg_id)
if reg is None or reg["status"] != "approved":
raise HTTPException(status_code=401, detail="Unauthorized")
user_identifier = reg["login"]
user_name = reg["login"]
else:
# Legacy api_key auth
if not body.user_id:
raise HTTPException(status_code=401, detail="Unauthorized")
key_hash = _hash_api_key(credentials.credentials)
stored_hash = await db.get_api_key_hash_by_uuid(body.user_id)
if stored_hash is None or not secrets.compare_digest(key_hash, stored_hash):
raise HTTPException(status_code=401, detail="Unauthorized")
if await db.is_user_blocked(body.user_id):
raise HTTPException(status_code=403, detail="User is blocked")
user_identifier = body.user_id
user_name = await db.get_user_name(body.user_id) or body.user_id[:8]
geo = body.geo
lat = geo.lat if geo else None
@ -160,43 +216,65 @@ async def signal(
accuracy = geo.accuracy if geo else None
signal_id = await db.save_signal(
user_uuid=body.user_id,
user_uuid=user_identifier,
timestamp=body.timestamp,
lat=lat,
lon=lon,
accuracy=accuracy,
)
user_name = await db.get_user_name(body.user_id)
ts = datetime.fromtimestamp(body.timestamp / 1000, tz=timezone.utc)
name = user_name or body.user_id[:8]
geo_info = (
f"📍 {lat}, {lon}{accuracy}м)"
f"📍 <a href=\"https://maps.google.com/maps?q={lat},{lon}\">{lat}, {lon}</a>{accuracy:.0f}м)"
if geo
else "Без геолокации"
)
text = (
f"🚨 Сигнал от {name}\n"
f"{ts.strftime('%H:%M:%S')} UTC\n"
f"{geo_info}"
else "Гео нету"
)
if body.is_test:
text = (
f"🧪 Тест от {user_name}\n"
f"{ts.strftime('%H:%M:%S')} UTC\n"
f"{geo_info}"
)
else:
text = (
f"🚨 Сигнал от {user_name}\n"
f"{ts.strftime('%H:%M:%S')} UTC\n"
f"{geo_info}"
)
asyncio.create_task(telegram.send_message(text))
return SignalResponse(status="ok", signal_id=signal_id)
_ALLOWED_EMAIL_DOMAIN = "tutlot.com"
_VIOLATION_BLOCK_THRESHOLD = 5
@app.post("/api/auth/register", response_model=AuthRegisterResponse, status_code=201)
async def auth_register(
request: Request,
body: AuthRegisterRequest,
_: None = Depends(rate_limit_auth_register),
__: None = Depends(check_ip_not_blocked),
) -> AuthRegisterResponse:
# Domain verification (server-side only)
email_str = str(body.email)
domain = email_str.rsplit("@", 1)[-1].lower() if "@" in email_str else ""
if domain != _ALLOWED_EMAIL_DOMAIN:
client_ip = _get_client_ip(request)
count = await db.record_ip_violation(client_ip)
logger.warning("Domain violation from %s (attempt %d): %s", client_ip, count, email_str)
raise HTTPException(
status_code=403,
detail="Ваш IP отправлен компетентным службам и за вами уже выехали. Ожидайте.",
)
password_hash = _hash_password(body.password)
push_sub_json = (
body.push_subscription.model_dump_json() if body.push_subscription else None
)
try:
reg_id = await db.create_registration(
email=str(body.email),
email=email_str,
login=body.login,
password_hash=password_hash,
push_subscription=push_sub_json,
@ -211,13 +289,32 @@ async def auth_register(
telegram.send_registration_notification(
reg_id=reg_id,
login=body.login,
email=str(body.email),
email=email_str,
created_at=reg["created_at"] if reg else "",
)
)
return AuthRegisterResponse(status="pending", message="Заявка отправлена на рассмотрение")
@app.post("/api/auth/login", response_model=AuthLoginResponse)
async def auth_login(
body: AuthLoginRequest,
_: None = Depends(rate_limit_auth_login),
__: None = Depends(check_ip_not_blocked),
) -> AuthLoginResponse:
reg = await db.get_registration_by_login_or_email(body.login_or_email)
if reg is None or not _verify_password(body.password, reg["password_hash"]):
raise HTTPException(status_code=401, detail="Неверный логин или пароль")
if reg["status"] == "pending":
raise HTTPException(status_code=403, detail="Ваша заявка ожидает рассмотрения")
if reg["status"] == "rejected":
raise HTTPException(status_code=403, detail="Ваша заявка отклонена")
if reg["status"] != "approved":
raise HTTPException(status_code=403, detail="Доступ запрещён")
token = create_auth_token(reg["id"], reg["login"])
return AuthLoginResponse(token=token, login=reg["login"])
async def _handle_callback_query(cb: dict) -> None:
"""Process approve/reject callback from admin Telegram inline buttons."""
data = cb.get("data", "")
@ -240,7 +337,11 @@ async def _handle_callback_query(cb: dict) -> None:
return
if action == "approve":
await db.update_registration_status(reg_id, "approved")
updated = await db.update_registration_status(reg_id, "approved")
if not updated:
# Already processed (not pending) — ack the callback and stop
await telegram.answer_callback_query(callback_query_id)
return
if chat_id and message_id:
await telegram.edit_message_text(
chat_id, message_id, f"✅ Пользователь {reg['login']} одобрен"
@ -254,7 +355,10 @@ async def _handle_callback_query(cb: dict) -> None:
)
)
elif action == "reject":
await db.update_registration_status(reg_id, "rejected")
updated = await db.update_registration_status(reg_id, "rejected")
if not updated:
await telegram.answer_callback_query(callback_query_id)
return
if chat_id and message_id:
await telegram.edit_message_text(
chat_id, message_id, f"❌ Пользователь {reg['login']} отклонён"

View file

@ -1,6 +1,11 @@
from __future__ import annotations
import base64
import hashlib
import hmac
import json
import secrets
import time
from typing import Optional
from fastapi import Depends, Header, HTTPException, Request
@ -8,6 +13,12 @@ from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from backend import config, db
# JWT secret: stable across restarts if JWT_SECRET env var is set; random per-process otherwise
_JWT_SECRET: str = config.JWT_SECRET or secrets.token_hex(32)
_JWT_HEADER_B64: str = (
base64.urlsafe_b64encode(b'{"alg":"HS256","typ":"JWT"}').rstrip(b"=").decode()
)
_bearer = HTTPBearer(auto_error=False)
_RATE_LIMIT = 5
@ -28,6 +39,12 @@ def _get_client_ip(request: Request) -> str:
)
async def check_ip_not_blocked(request: Request) -> None:
ip = _get_client_ip(request)
if await db.is_ip_blocked(ip):
raise HTTPException(status_code=403, detail="Доступ запрещён")
async def verify_webhook_secret(
x_telegram_bot_api_secret_token: str = Header(default=""),
) -> None:
@ -65,3 +82,74 @@ async def rate_limit_auth_register(request: Request) -> None:
count = await db.rate_limit_increment(key, _AUTH_REGISTER_RATE_WINDOW)
if count > _AUTH_REGISTER_RATE_LIMIT:
raise HTTPException(status_code=429, detail="Too Many Requests")
_AUTH_LOGIN_RATE_LIMIT = 5
_AUTH_LOGIN_RATE_WINDOW = 900 # 15 minutes
def _b64url_encode(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
def _b64url_decode(s: str) -> bytes:
padding = 4 - len(s) % 4
if padding != 4:
s += "=" * padding
return base64.urlsafe_b64decode(s)
def create_auth_token(reg_id: int, login: str) -> str:
"""Create a signed HS256 JWT for an approved registration."""
now = int(time.time())
payload = {
"sub": str(reg_id),
"login": login,
"iat": now,
"exp": now + config.JWT_TOKEN_EXPIRE_SECONDS,
}
payload_b64 = _b64url_encode(json.dumps(payload, separators=(",", ":")).encode())
signing_input = f"{_JWT_HEADER_B64}.{payload_b64}"
sig = hmac.new(
_JWT_SECRET.encode(), signing_input.encode(), hashlib.sha256
).digest()
return f"{signing_input}.{_b64url_encode(sig)}"
def _verify_jwt_token(token: str) -> dict:
"""Verify token signature and expiry. Returns payload dict on success."""
parts = token.split(".")
if len(parts) != 3:
raise ValueError("Invalid token format")
header_b64, payload_b64, sig_b64 = parts
signing_input = f"{header_b64}.{payload_b64}"
expected_sig = hmac.new(
_JWT_SECRET.encode(), signing_input.encode(), hashlib.sha256
).digest()
actual_sig = _b64url_decode(sig_b64)
if not hmac.compare_digest(expected_sig, actual_sig):
raise ValueError("Invalid signature")
payload = json.loads(_b64url_decode(payload_b64))
if payload.get("exp", 0) < time.time():
raise ValueError("Token expired")
return payload
async def rate_limit_auth_login(request: Request) -> None:
key = f"login:{_get_client_ip(request)}"
count = await db.rate_limit_increment(key, _AUTH_LOGIN_RATE_WINDOW)
if count > _AUTH_LOGIN_RATE_LIMIT:
raise HTTPException(status_code=429, detail="Too Many Requests")
async def verify_auth_token(
credentials: Optional[HTTPAuthorizationCredentials] = Depends(_bearer),
) -> dict:
"""Dependency for protected endpoints — verifies Bearer JWT, returns payload."""
if credentials is None:
raise HTTPException(status_code=401, detail="Unauthorized")
try:
payload = _verify_jwt_token(credentials.credentials)
except Exception:
raise HTTPException(status_code=401, detail="Unauthorized")
return payload

View file

@ -22,9 +22,10 @@ class GeoData(BaseModel):
class SignalRequest(BaseModel):
user_id: str = Field(..., pattern=r'^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$')
user_id: Optional[str] = None # UUID for legacy api_key auth; omit for JWT auth
timestamp: int = Field(..., gt=0)
geo: Optional[GeoData] = None
is_test: bool = False
class SignalResponse(BaseModel):
@ -66,3 +67,13 @@ class AuthRegisterRequest(BaseModel):
class AuthRegisterResponse(BaseModel):
status: str
message: str
class AuthLoginRequest(BaseModel):
login_or_email: str = Field(..., min_length=1, max_length=255)
password: str = Field(..., min_length=1, max_length=128)
class AuthLoginResponse(BaseModel):
token: str
login: str

View file

@ -11,11 +11,6 @@ from backend import config, db
logger = logging.getLogger(__name__)
# Suppress httpx/httpcore transport-level logging to prevent BOT_TOKEN URL leakage.
# httpx logs request URLs (which embed the token) at DEBUG/INFO level depending on version.
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
_TELEGRAM_API = "https://api.telegram.org/bot{token}/{method}"
@ -55,7 +50,7 @@ async def send_message(text: str) -> None:
url = _TELEGRAM_API.format(token=config.BOT_TOKEN, method="sendMessage")
async with httpx.AsyncClient(timeout=10) as client:
for attempt in range(3):
resp = await client.post(url, json={"chat_id": config.CHAT_ID, "text": text})
resp = await client.post(url, json={"chat_id": config.CHAT_ID, "text": text, "parse_mode": "HTML"})
if resp.status_code == 429:
retry_after = resp.json().get("parameters", {}).get("retry_after", 30)
sleep = retry_after * (attempt + 1)
@ -106,7 +101,8 @@ async def send_registration_notification(
resp.text,
)
except Exception as exc:
logger.error("send_registration_notification error: %s", exc)
# Do not log exc directly — httpx exceptions embed the full API URL with BOT_TOKEN
logger.error("send_registration_notification error: %s", type(exc).__name__)
async def answer_callback_query(callback_query_id: str) -> None:
@ -118,7 +114,8 @@ async def answer_callback_query(callback_query_id: str) -> None:
if resp.status_code != 200:
logger.error("answerCallbackQuery failed %s: %s", resp.status_code, resp.text)
except Exception as exc:
logger.error("answerCallbackQuery error: %s", exc)
# Do not log exc directly — httpx exceptions embed the full API URL with BOT_TOKEN
logger.error("answerCallbackQuery error: %s", type(exc).__name__)
async def edit_message_text(chat_id: str | int, message_id: int, text: str) -> None:
@ -132,7 +129,8 @@ async def edit_message_text(chat_id: str | int, message_id: int, text: str) -> N
if resp.status_code != 200:
logger.error("editMessageText failed %s: %s", resp.status_code, resp.text)
except Exception as exc:
logger.error("editMessageText error: %s", exc)
# Do not log exc directly — httpx exceptions embed the full API URL with BOT_TOKEN
logger.error("editMessageText error: %s", type(exc).__name__)
async def set_webhook(url: str, secret: str) -> None:
@ -145,76 +143,3 @@ async def set_webhook(url: str, secret: str) -> None:
raise RuntimeError(f"setWebhook failed: {resp.text}")
logger.info("Webhook registered: %s", url)
# v2.0 feature
class SignalAggregator:
def __init__(self, interval: int = 10) -> None:
self._interval = interval
self._buffer: list[dict] = []
self._lock = asyncio.Lock()
self._stopped = False
async def add_signal(
self,
user_uuid: str,
user_name: Optional[str],
timestamp: int,
geo: Optional[dict],
signal_id: int,
) -> None:
async with self._lock:
self._buffer.append(
{
"user_uuid": user_uuid,
"user_name": user_name,
"timestamp": timestamp,
"geo": geo,
"signal_id": signal_id,
}
)
async def flush(self) -> None:
async with self._lock:
if not self._buffer:
return
items = self._buffer[:]
self._buffer.clear()
signal_ids = [item["signal_id"] for item in items]
timestamps = [item["timestamp"] for item in items]
ts_start = datetime.fromtimestamp(min(timestamps) / 1000, tz=timezone.utc)
ts_end = datetime.fromtimestamp(max(timestamps) / 1000, tz=timezone.utc)
t_fmt = "%H:%M:%S"
names = []
for item in items:
name = item["user_name"]
label = name if name else item["user_uuid"][:8]
names.append(label)
geo_count = sum(1 for item in items if item["geo"])
n = len(items)
text = (
f"\U0001f6a8 Получено {n} сигнал{'ов' if n != 1 else ''} "
f"[{ts_start.strftime(t_fmt)}{ts_end.strftime(t_fmt)}]\n"
f"Пользователи: {', '.join(names)}\n"
f"\U0001f4cd С геолокацией: {geo_count} из {n}"
)
try:
await send_message(text)
await db.save_telegram_batch(text, n, signal_ids)
# rate-limit: 1 msg/sec max (#1014)
await asyncio.sleep(1)
except Exception:
logger.exception("Failed to flush aggregator batch")
async def run(self) -> None:
while not self._stopped:
await asyncio.sleep(self._interval)
if self._buffer:
await self.flush()
def stop(self) -> None:
self._stopped = True

View file

@ -8,6 +8,7 @@ Type=simple
User=www-data
WorkingDirectory=/opt/baton
EnvironmentFile=/opt/baton/.env
ExecStartPre=/opt/baton/venv/bin/pip install -r /opt/baton/requirements.txt -q
ExecStart=/opt/baton/venv/bin/uvicorn backend.main:app --host 127.0.0.1 --port 8000
Restart=on-failure
RestartSec=5s

23
docker-compose.yml Normal file
View file

@ -0,0 +1,23 @@
services:
backend:
build: .
restart: unless-stopped
env_file: .env
environment:
DB_PATH: /data/baton.db
volumes:
- db_data:/data
nginx:
image: nginx:alpine
restart: unless-stopped
ports:
- "127.0.0.1:8080:80"
volumes:
- ./frontend:/usr/share/nginx/html:ro
- ./nginx/docker.conf:/etc/nginx/conf.d/default.conf:ro
depends_on:
- backend
volumes:
db_data:

View file

@ -39,31 +39,26 @@ function _initStorage() {
// ========== User identity ==========
function _getOrCreateUserId() {
let id = _storage.getItem('baton_user_id');
if (!id) {
id = crypto.randomUUID();
_storage.setItem('baton_user_id', id);
}
return id;
}
function _isRegistered() {
return _storage.getItem('baton_registered') === '1';
return !!_storage.getItem('baton_auth_token');
}
function _getUserName() {
return _storage.getItem('baton_user_name') || '';
return _storage.getItem('baton_login') || '';
}
function _getApiKey() {
return _storage.getItem('baton_api_key') || '';
function _getAuthToken() {
return _storage.getItem('baton_auth_token') || '';
}
function _saveRegistration(name, apiKey) {
_storage.setItem('baton_user_name', name);
_storage.setItem('baton_registered', '1');
if (apiKey) _storage.setItem('baton_api_key', apiKey);
function _saveAuth(token, login) {
_storage.setItem('baton_auth_token', token);
_storage.setItem('baton_login', login);
}
function _clearAuth() {
_storage.removeItem('baton_auth_token');
_storage.removeItem('baton_login');
}
function _getInitials(name) {
@ -92,6 +87,29 @@ function _setStatus(msg, cls) {
el.hidden = !msg;
}
function _setRegStatus(msg, cls) {
const el = document.getElementById('reg-status');
if (!el) return;
el.textContent = msg;
el.className = 'reg-status' + (cls ? ' reg-status--' + cls : '');
el.hidden = !msg;
}
function _setLoginStatus(msg, cls) {
const el = document.getElementById('login-status');
if (!el) return;
el.textContent = msg;
el.className = 'reg-status' + (cls ? ' reg-status--' + cls : '');
el.hidden = !msg;
}
function _showView(id) {
['view-login', 'view-register'].forEach((vid) => {
const el = document.getElementById(vid);
if (el) el.hidden = vid !== id;
});
}
function _updateNetworkIndicator() {
const el = document.getElementById('indicator-network');
if (!el) return;
@ -142,23 +160,38 @@ function _getGeo() {
// ========== Handlers ==========
async function _handleRegister() {
const input = document.getElementById('name-input');
const btn = document.getElementById('btn-confirm');
const name = input.value.trim();
if (!name) return;
async function _handleLogin() {
const loginInput = document.getElementById('login-input');
const passInput = document.getElementById('login-password');
const btn = document.getElementById('btn-login');
const login = loginInput.value.trim();
const password = passInput.value;
if (!login || !password) return;
btn.disabled = true;
_setStatus('', '');
_setLoginStatus('', '');
try {
const uuid = _getOrCreateUserId();
const data = await _apiPost('/api/register', { uuid, name });
_saveRegistration(name, data.api_key);
const data = await _apiPost('/api/auth/login', {
login_or_email: login,
password: password,
});
_saveAuth(data.token, data.login);
passInput.value = '';
_updateUserAvatar();
_showMain();
} catch (_) {
_setStatus('Error. Please try again.', 'error');
} catch (err) {
let msg = 'Ошибка входа. Попробуйте ещё раз.';
if (err && err.message) {
const colonIdx = err.message.indexOf(': ');
if (colonIdx !== -1) {
try {
const parsed = JSON.parse(err.message.slice(colonIdx + 2));
if (parsed.detail) msg = parsed.detail;
} catch (_) {}
}
}
_setLoginStatus(msg, 'error');
btn.disabled = false;
}
}
@ -170,10 +203,43 @@ function _setSosState(state) {
btn.disabled = state === 'sending';
}
async function _handleSignal() {
// v1: no offline queue — show error and return (decision #1019)
async function _handleTestSignal() {
if (!navigator.onLine) {
_setStatus('No connection. Check your network and try again.', 'error');
_setStatus('Нет соединения.', 'error');
return;
}
const token = _getAuthToken();
if (!token) return;
_setStatus('', '');
try {
const geo = await _getGeo();
const body = { timestamp: Date.now(), is_test: true };
if (geo) body.geo = geo;
await _apiPost('/api/signal', body, { Authorization: 'Bearer ' + token });
_setStatus('Тест отправлен', 'success');
setTimeout(() => _setStatus('', ''), 1500);
} catch (err) {
if (err && err.status === 401) {
_clearAuth();
_setStatus('Сессия истекла. Войдите заново.', 'error');
setTimeout(() => _showOnboarding(), 1500);
} else {
_setStatus('Ошибка отправки.', 'error');
}
}
}
async function _handleSignal() {
if (!navigator.onLine) {
_setStatus('Нет соединения. Проверьте сеть и попробуйте снова.', 'error');
return;
}
const token = _getAuthToken();
if (!token) {
_clearAuth();
_showOnboarding();
return;
}
@ -182,16 +248,13 @@ async function _handleSignal() {
try {
const geo = await _getGeo();
const uuid = _getOrCreateUserId();
const body = { user_id: uuid, timestamp: Date.now() };
const body = { timestamp: Date.now() };
if (geo) body.geo = geo;
const apiKey = _getApiKey();
const authHeaders = apiKey ? { Authorization: 'Bearer ' + apiKey } : {};
await _apiPost('/api/signal', body, authHeaders);
await _apiPost('/api/signal', body, { Authorization: 'Bearer ' + token });
_setSosState('success');
_setStatus('Signal sent!', 'success');
_setStatus('Сигнал отправлен!', 'success');
setTimeout(() => {
_setSosState('default');
_setStatus('', '');
@ -199,9 +262,11 @@ async function _handleSignal() {
} catch (err) {
_setSosState('default');
if (err && err.status === 401) {
_setStatus('Session expired or key is invalid. Please re-register.', 'error');
_clearAuth();
_setStatus('Сессия истекла. Войдите заново.', 'error');
setTimeout(() => _showOnboarding(), 1500);
} else {
_setStatus('Error sending. Try again.', 'error');
_setStatus('Ошибка отправки. Попробуйте ещё раз.', 'error');
}
}
}
@ -210,17 +275,44 @@ async function _handleSignal() {
function _showOnboarding() {
_showScreen('screen-onboarding');
_showView('view-login');
const input = document.getElementById('name-input');
const btn = document.getElementById('btn-confirm');
const loginInput = document.getElementById('login-input');
const passInput = document.getElementById('login-password');
const btnLogin = document.getElementById('btn-login');
input.addEventListener('input', () => {
btn.disabled = input.value.trim().length === 0;
function _updateLoginBtn() {
btnLogin.disabled = !loginInput.value.trim() || !passInput.value;
}
loginInput.addEventListener('input', _updateLoginBtn);
passInput.addEventListener('input', _updateLoginBtn);
passInput.addEventListener('keydown', (e) => {
if (e.key === 'Enter' && !btnLogin.disabled) _handleLogin();
});
input.addEventListener('keydown', (e) => {
if (e.key === 'Enter' && !btn.disabled) _handleRegister();
});
btn.addEventListener('click', _handleRegister);
btnLogin.addEventListener('click', _handleLogin);
const btnToRegister = document.getElementById('btn-switch-to-register');
if (btnToRegister) {
btnToRegister.addEventListener('click', () => {
_setRegStatus('', '');
_setLoginStatus('', '');
_showView('view-register');
});
}
const btnToLogin = document.getElementById('btn-switch-to-login');
if (btnToLogin) {
btnToLogin.addEventListener('click', () => {
_setLoginStatus('', '');
_showView('view-login');
});
}
const btnRegister = document.getElementById('btn-register');
if (btnRegister) {
btnRegister.addEventListener('click', _handleSignUp);
}
}
function _showMain() {
@ -232,6 +324,20 @@ function _showMain() {
btn.addEventListener('click', _handleSignal);
btn.dataset.listenerAttached = '1';
}
// Avatar and network indicator → test signal (only on main screen)
const avatar = document.getElementById('user-avatar');
if (avatar && !avatar.dataset.testAttached) {
avatar.addEventListener('click', _handleTestSignal);
avatar.dataset.testAttached = '1';
avatar.style.cursor = 'pointer';
}
const indicator = document.getElementById('indicator-network');
if (indicator && !indicator.dataset.testAttached) {
indicator.addEventListener('click', _handleTestSignal);
indicator.dataset.testAttached = '1';
indicator.style.cursor = 'pointer';
}
}
// ========== Service Worker ==========
@ -243,16 +349,150 @@ function _registerSW() {
});
}
// ========== VAPID / Push subscription ==========
async function _fetchVapidPublicKey() {
try {
const res = await fetch('/api/push/public-key');
if (!res.ok) {
console.warn('[baton] /api/push/public-key returned', res.status);
return null;
}
const data = await res.json();
return data.vapid_public_key || null;
} catch (err) {
console.warn('[baton] Failed to fetch VAPID public key:', err);
return null;
}
}
function _urlBase64ToUint8Array(base64String) {
const padding = '='.repeat((4 - (base64String.length % 4)) % 4);
const base64 = (base64String + padding).replace(/-/g, '+').replace(/_/g, '/');
const raw = atob(base64);
const output = new Uint8Array(raw.length);
for (let i = 0; i < raw.length; i++) {
output[i] = raw.charCodeAt(i);
}
return output;
}
async function _initPushSubscription(vapidPublicKey) {
if (!vapidPublicKey) {
console.warn('[baton] VAPID public key not available — push subscription skipped');
return;
}
if (!('serviceWorker' in navigator) || !('PushManager' in window)) {
return;
}
try {
const registration = await navigator.serviceWorker.ready;
const existing = await registration.pushManager.getSubscription();
if (existing) return;
const applicationServerKey = _urlBase64ToUint8Array(vapidPublicKey);
const subscription = await registration.pushManager.subscribe({
userVisibleOnly: true,
applicationServerKey,
});
_storage.setItem('baton_push_subscription', JSON.stringify(subscription));
console.info('[baton] Push subscription created');
} catch (err) {
console.warn('[baton] Push subscription failed:', err);
}
}
// ========== Registration (account sign-up) ==========
async function _getPushSubscriptionForReg() {
if (!('serviceWorker' in navigator) || !('PushManager' in window)) return null;
try {
const vapidKey = await _fetchVapidPublicKey();
if (!vapidKey) return null;
const registration = await navigator.serviceWorker.ready;
const existing = await registration.pushManager.getSubscription();
if (existing) return existing.toJSON();
const applicationServerKey = _urlBase64ToUint8Array(vapidKey);
const subscription = await registration.pushManager.subscribe({
userVisibleOnly: true,
applicationServerKey,
});
return subscription.toJSON();
} catch (err) {
console.warn('[baton] Push subscription for registration failed:', err);
return null;
}
}
async function _handleSignUp() {
const emailInput = document.getElementById('reg-email');
const loginInput = document.getElementById('reg-login');
const passwordInput = document.getElementById('reg-password');
const btn = document.getElementById('btn-register');
if (!emailInput || !loginInput || !passwordInput || !btn) return;
const email = emailInput.value.trim();
const login = loginInput.value.trim();
const password = passwordInput.value;
if (!email || !login || !password) {
_setRegStatus('Заполните все поля.', 'error');
return;
}
if (!/^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email)) {
_setRegStatus('Введите корректный email.', 'error');
return;
}
btn.disabled = true;
const originalText = btn.textContent.trim();
btn.textContent = '...';
_setRegStatus('', '');
try {
const push_subscription = await _getPushSubscriptionForReg().catch(() => null);
await _apiPost('/api/auth/register', { email, login, password, push_subscription });
passwordInput.value = '';
_setRegStatus('Заявка отправлена. Ожидайте подтверждения администратора.', 'success');
} catch (err) {
let msg = 'Ошибка. Попробуйте ещё раз.';
if (err && err.message) {
const colonIdx = err.message.indexOf(': ');
if (colonIdx !== -1) {
try {
const parsed = JSON.parse(err.message.slice(colonIdx + 2));
if (parsed.detail) msg = parsed.detail;
} catch (_) {}
}
}
if (err && err.status === 403 && msg !== 'Ошибка. Попробуйте ещё раз.') {
_showBlockScreen(msg);
} else {
_setRegStatus(msg, 'error');
btn.disabled = false;
btn.textContent = originalText;
}
}
}
function _showBlockScreen(msg) {
const screen = document.getElementById('screen-onboarding');
if (!screen) return;
screen.innerHTML =
'<div class="screen-content">' +
'<p class="block-message">' + msg + '</p>' +
'<button type="button" class="btn-confirm" id="btn-block-ok">OK</button>' +
'</div>';
document.getElementById('btn-block-ok').addEventListener('click', () => {
location.reload();
});
}
// ========== Init ==========
function _init() {
_initStorage();
// Pre-generate and persist UUID on first visit (per arch spec flow)
_getOrCreateUserId();
// Private mode graceful degradation (decision #1041):
// show inline banner with explicit action guidance when localStorage is unavailable
// Private mode graceful degradation (decision #1041)
if (_storageType !== 'local') {
const banner = document.getElementById('private-mode-banner');
if (banner) banner.hidden = false;
@ -263,12 +503,17 @@ function _init() {
window.addEventListener('online', _updateNetworkIndicator);
window.addEventListener('offline', _updateNetworkIndicator);
// Route to correct screen
// Route to correct screen based on JWT token presence
if (_isRegistered()) {
_showMain();
} else {
_showOnboarding();
}
// Fire-and-forget: fetch VAPID key from API and subscribe to push (non-blocking)
_fetchVapidPublicKey().then(_initPushSubscription).catch((err) => {
console.warn('[baton] Push init error:', err);
});
}
document.addEventListener('DOMContentLoaded', () => {

View file

@ -2,7 +2,7 @@
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1, viewport-fit=cover">
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1, user-scalable=no, viewport-fit=cover">
<meta name="description" content="Emergency signal button">
<!-- PWA meta tags -->
@ -36,23 +36,80 @@
<!-- Onboarding screen: shown on first visit (no UUID registered yet) -->
<div id="screen-onboarding" class="screen" role="main" hidden>
<div class="screen-content">
<!-- View: login with credentials -->
<div class="screen-content" id="view-login">
<input
type="text"
id="name-input"
id="login-input"
class="name-input"
placeholder="Your name"
maxlength="100"
autocomplete="name"
placeholder="Логин или email"
maxlength="255"
autocomplete="username"
autocorrect="off"
autocapitalize="words"
autocapitalize="none"
spellcheck="false"
aria-label="Your name"
aria-label="Логин или email"
>
<button type="button" id="btn-confirm" class="btn-confirm" disabled>
Confirm
<input
type="password"
id="login-password"
class="name-input"
placeholder="Пароль"
autocomplete="current-password"
aria-label="Пароль"
>
<button type="button" id="btn-login" class="btn-confirm" disabled>
Войти
</button>
<div id="login-status" class="reg-status" hidden></div>
<button type="button" id="btn-switch-to-register" class="btn-link">
Зарегистрироваться
</button>
</div>
<!-- View: account registration -->
<div class="screen-content" id="view-register" hidden>
<input
type="email"
id="reg-email"
class="name-input"
placeholder="Email"
autocomplete="email"
autocorrect="off"
autocapitalize="none"
spellcheck="false"
aria-label="Email"
>
<input
type="text"
id="reg-login"
class="name-input"
placeholder="Логин"
maxlength="64"
autocomplete="username"
autocorrect="off"
autocapitalize="none"
spellcheck="false"
aria-label="Логин"
>
<input
type="password"
id="reg-password"
class="name-input"
placeholder="Пароль"
autocomplete="new-password"
aria-label="Пароль"
>
<button type="button" id="btn-register" class="btn-confirm">
Зарегистрироваться
</button>
<button type="button" id="btn-switch-to-login" class="btn-link">
← Назад
</button>
<div id="reg-status" class="reg-status" hidden></div>
</div>
</div>
<!-- Main screen: SOS button -->

View file

@ -28,14 +28,14 @@ html, body {
-webkit-tap-highlight-color: transparent;
overscroll-behavior: none;
user-select: none;
overflow: hidden;
}
body {
display: flex;
flex-direction: column;
min-height: 100vh;
/* Use dynamic viewport height on mobile to account for browser chrome */
min-height: 100dvh;
height: 100vh;
height: 100dvh;
}
/* ===== Private mode banner (decision #1041) ===== */
@ -59,6 +59,7 @@ body {
justify-content: space-between;
align-items: center;
padding: 16px 20px;
padding-top: calc(env(safe-area-inset-top, 0px) + 16px);
flex-shrink: 0;
}
@ -148,10 +149,8 @@ body {
/* ===== SOS button (min 60vmin × 60vmin per UX spec) ===== */
.btn-sos {
width: 60vmin;
height: 60vmin;
min-width: 180px;
min-height: 180px;
width: min(60vmin, 70vw, 300px);
height: min(60vmin, 70vw, 300px);
border-radius: 50%;
border: none;
background: var(--sos);
@ -198,3 +197,44 @@ body {
.status[hidden] { display: none; }
.status--error { color: #f87171; }
.status--success { color: #4ade80; }
/* ===== Registration form ===== */
/* Override display:flex so [hidden] works on screen-content divs */
.screen-content[hidden] { display: none; }
.btn-link {
background: none;
border: none;
color: var(--muted);
font-size: 14px;
cursor: pointer;
padding: 4px 0;
text-decoration: underline;
text-underline-offset: 2px;
-webkit-tap-highlight-color: transparent;
}
.btn-link:active { color: var(--text); }
.reg-status {
width: 100%;
max-width: 320px;
font-size: 14px;
text-align: center;
line-height: 1.5;
padding: 4px 0;
}
.reg-status[hidden] { display: none; }
.reg-status--error { color: #f87171; }
.reg-status--success { color: #4ade80; }
.block-message {
color: #f87171;
font-size: 16px;
text-align: center;
line-height: 1.6;
padding: 20px;
max-width: 320px;
}

View file

@ -1,6 +1,6 @@
'use strict';
const CACHE_NAME = 'baton-v1';
const CACHE_NAME = 'baton-v4';
// App shell assets to precache
const APP_SHELL = [

61
nginx/docker.conf Normal file
View file

@ -0,0 +1,61 @@
map $request_uri $masked_uri {
default $request_uri;
"~^(/bot)[^/]+(/.*)$" "$1[REDACTED]$2";
}
log_format baton_secure '$remote_addr - $remote_user [$time_local] '
'"$request_method $masked_uri $server_protocol" '
'$status $body_bytes_sent '
'"$http_referer" "$http_user_agent"';
server {
listen 80;
server_name _;
access_log /var/log/nginx/baton_access.log baton_secure;
error_log /var/log/nginx/baton_error.log warn;
add_header X-Content-Type-Options nosniff always;
add_header X-Frame-Options DENY always;
add_header Content-Security-Policy "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; connect-src 'self'" always;
# API + health + admin → backend
location /api/ {
proxy_pass http://backend:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_read_timeout 30s;
proxy_send_timeout 30s;
proxy_connect_timeout 5s;
}
location /health {
proxy_pass http://backend:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
location /admin/users {
proxy_pass http://backend:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_read_timeout 30s;
}
# Frontend static
location / {
root /usr/share/nginx/html;
try_files $uri /index.html;
expires 1h;
add_header Cache-Control "public" always;
add_header X-Content-Type-Options nosniff always;
add_header X-Frame-Options DENY always;
add_header Content-Security-Policy "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; connect-src 'self'" always;
}
}

View file

@ -22,6 +22,7 @@ os.environ.setdefault("WEBHOOK_SECRET", "test-webhook-secret")
os.environ.setdefault("WEBHOOK_URL", "https://example.com/api/webhook/telegram")
os.environ.setdefault("FRONTEND_ORIGIN", "http://localhost:3000")
os.environ.setdefault("ADMIN_TOKEN", "test-admin-token")
os.environ.setdefault("ADMIN_CHAT_ID", "5694335584")
# ── 2. aiosqlite monkey-patch ────────────────────────────────────────────────
import aiosqlite
@ -69,14 +70,20 @@ def temp_db():
# ── 5. App client factory ────────────────────────────────────────────────────
def make_app_client():
def make_app_client(capture_send_requests: list | None = None):
"""
Async context manager that:
1. Assigns a fresh temp-file DB path
2. Mocks Telegram setWebhook, sendMessage, answerCallbackQuery, editMessageText
3. Runs the FastAPI lifespan (startup test shutdown)
4. Yields an httpx.AsyncClient wired to the app
Args:
capture_send_requests: if provided, each sendMessage request body (dict) is
appended to this list, enabling HTTP-level assertions on chat_id, text, etc.
"""
import json as _json
tg_set_url = f"https://api.telegram.org/bot{config.BOT_TOKEN}/setWebhook"
send_url = f"https://api.telegram.org/bot{config.BOT_TOKEN}/sendMessage"
get_me_url = f"https://api.telegram.org/bot{config.BOT_TOKEN}/getMe"
@ -95,9 +102,18 @@ def make_app_client():
mock_router.post(tg_set_url).mock(
return_value=httpx.Response(200, json={"ok": True, "result": True})
)
mock_router.post(send_url).mock(
return_value=httpx.Response(200, json={"ok": True})
)
if capture_send_requests is not None:
def _capture_send(request: httpx.Request) -> httpx.Response:
try:
capture_send_requests.append(_json.loads(request.content))
except Exception:
pass
return httpx.Response(200, json={"ok": True})
mock_router.post(send_url).mock(side_effect=_capture_send)
else:
mock_router.post(send_url).mock(
return_value=httpx.Response(200, json={"ok": True})
)
mock_router.post(answer_cb_url).mock(
return_value=httpx.Response(200, json={"ok": True})
)

View file

@ -119,7 +119,7 @@ async def test_signal_message_contains_registered_username():
@pytest.mark.asyncio
async def test_signal_message_without_geo_contains_bez_geolocatsii():
"""When geo is None, message must contain 'Без геолокации'."""
"""When geo is None, message must contain 'Гео нету'."""
async with make_app_client() as client:
api_key = await _register(client, _UUID_S3, "Bob")
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
@ -129,7 +129,7 @@ async def test_signal_message_without_geo_contains_bez_geolocatsii():
headers={"Authorization": f"Bearer {api_key}"},
)
text = mock_send.call_args[0][0]
assert "Без геолокации" in text
assert "Гео нету" in text
@pytest.mark.asyncio
@ -168,25 +168,17 @@ async def test_signal_message_contains_utc_marker():
# ---------------------------------------------------------------------------
# Criterion 3 — SignalAggregator preserved with '# v2.0 feature' marker (static)
# Criterion 3 — SignalAggregator removed (BATON-BIZ-004: dead code cleanup)
# ---------------------------------------------------------------------------
def test_signal_aggregator_class_preserved_in_telegram():
"""SignalAggregator class must still exist in telegram.py (ADR-004, reserved for v2)."""
def test_signal_aggregator_class_removed_from_telegram():
"""SignalAggregator must be removed from telegram.py (BATON-BIZ-004)."""
source = (_BACKEND_DIR / "telegram.py").read_text()
assert "class SignalAggregator" in source
assert "class SignalAggregator" not in source
def test_signal_aggregator_has_v2_feature_comment():
"""The line immediately before 'class SignalAggregator' must contain '# v2.0 feature'."""
lines = (_BACKEND_DIR / "telegram.py").read_text().splitlines()
class_line_idx = next(
(i for i, line in enumerate(lines) if "class SignalAggregator" in line), None
)
assert class_line_idx is not None, "class SignalAggregator not found in telegram.py"
assert class_line_idx > 0, "SignalAggregator is on the first line — no preceding comment line"
preceding_line = lines[class_line_idx - 1]
assert "# v2.0 feature" in preceding_line, (
f"Expected '# v2.0 feature' on line before class SignalAggregator, got: {preceding_line!r}"
)
def test_signal_aggregator_not_referenced_in_telegram():
"""telegram.py must not reference SignalAggregator at all (BATON-BIZ-004)."""
source = (_BACKEND_DIR / "telegram.py").read_text()
assert "SignalAggregator" not in source

View file

@ -222,9 +222,9 @@ def test_html_loads_app_js() -> None:
assert "/app.js" in _html()
def test_html_has_name_input() -> None:
"""index.html must have name input field for onboarding."""
assert 'id="name-input"' in _html()
def test_html_has_login_input() -> None:
"""index.html must have login input field for onboarding."""
assert 'id="login-input"' in _html()
# ---------------------------------------------------------------------------
@ -316,31 +316,19 @@ def _app_js() -> str:
return (FRONTEND / "app.js").read_text(encoding="utf-8")
def test_app_uses_crypto_random_uuid() -> None:
"""app.js must generate UUID via crypto.randomUUID()."""
assert "crypto.randomUUID()" in _app_js()
def test_app_posts_to_auth_login() -> None:
"""app.js must send POST to /api/auth/login during login."""
assert "/api/auth/login" in _app_js()
def test_app_posts_to_api_register() -> None:
"""app.js must send POST to /api/register during onboarding."""
assert "/api/register" in _app_js()
def test_app_posts_to_auth_register() -> None:
"""app.js must send POST to /api/auth/register during registration."""
assert "/api/auth/register" in _app_js()
def test_app_register_sends_uuid() -> None:
"""app.js must include uuid in the /api/register request body."""
app = _app_js()
# The register call must include uuid in the payload
register_section = re.search(
r"_apiPost\(['\"]\/api\/register['\"].*?\)", app, re.DOTALL
)
assert register_section, "No _apiPost('/api/register') call found"
assert "uuid" in register_section.group(0), \
"uuid not included in /api/register call"
def test_app_uuid_saved_to_storage() -> None:
"""app.js must persist UUID to storage (baton_user_id key)."""
assert "baton_user_id" in _app_js()
def test_app_stores_auth_token() -> None:
"""app.js must persist JWT token to storage."""
assert "baton_auth_token" in _app_js()
assert "setItem" in _app_js()
@ -434,16 +422,14 @@ def test_app_posts_to_api_signal() -> None:
assert "/api/signal" in _app_js()
def test_app_signal_sends_user_id() -> None:
"""app.js must include user_id (UUID) in the /api/signal request body."""
def test_app_signal_sends_auth_header() -> None:
"""app.js must include Authorization Bearer header in /api/signal request."""
app = _app_js()
# The signal body may be built in a variable before passing to _apiPost
# Look for user_id key in the context around /api/signal
signal_area = re.search(
r"user_id.*?_apiPost\(['\"]\/api\/signal", app, re.DOTALL
r"_apiPost\(['\"]\/api\/signal['\"].*Authorization.*Bearer", app, re.DOTALL
)
assert signal_area, \
"user_id must be set in the request body before calling _apiPost('/api/signal')"
"Authorization Bearer header must be set in _apiPost('/api/signal') call"
def test_app_sos_button_click_calls_handle_signal() -> None:
@ -456,15 +442,15 @@ def test_app_sos_button_click_calls_handle_signal() -> None:
"btn-sos must be connected to _handleSignal"
def test_app_signal_uses_uuid_from_storage() -> None:
"""app.js must retrieve UUID from storage (_getOrCreateUserId) before sending signal."""
def test_app_signal_uses_token_from_storage() -> None:
"""app.js must retrieve auth token from storage before sending signal."""
app = _app_js()
handle_signal = re.search(
r"async function _handleSignal\(\).*?^}", app, re.MULTILINE | re.DOTALL
)
assert handle_signal, "_handleSignal function not found"
assert "_getOrCreateUserId" in handle_signal.group(0), \
"_handleSignal must call _getOrCreateUserId() to get UUID"
assert "_getAuthToken" in handle_signal.group(0), \
"_handleSignal must call _getAuthToken() to get JWT token"
# ---------------------------------------------------------------------------

View file

@ -140,7 +140,7 @@ async def test_signal_with_geo_send_message_contains_coordinates():
@pytest.mark.asyncio
async def test_signal_without_geo_send_message_contains_no_geo_label():
"""Criterion 1: when geo is null, Telegram message contains 'Без геолокации'."""
"""Criterion 1: when geo is null, Telegram message contains 'Гео нету'."""
sent_texts: list[str] = []
async def _capture(text: str) -> None:
@ -158,8 +158,8 @@ async def test_signal_without_geo_send_message_contains_no_geo_label():
await asyncio.sleep(0)
assert len(sent_texts) == 1
assert "Без геолокации" in sent_texts[0], (
f"Expected 'Без геолокации' in message, got: {sent_texts[0]!r}"
assert "Гео нету" in sent_texts[0], (
f"Expected 'Гео нету' in message, got: {sent_texts[0]!r}"
)

View file

@ -21,6 +21,7 @@ os.environ.setdefault("WEBHOOK_SECRET", "test-webhook-secret")
os.environ.setdefault("WEBHOOK_URL", "https://example.com/api/webhook/telegram")
os.environ.setdefault("FRONTEND_ORIGIN", "http://localhost:3000")
os.environ.setdefault("ADMIN_TOKEN", "test-admin-token")
os.environ.setdefault("ADMIN_CHAT_ID", "5694335584")
from unittest.mock import AsyncMock, patch
@ -32,7 +33,7 @@ _WEBHOOK_SECRET = "test-webhook-secret"
_WEBHOOK_HEADERS = {"X-Telegram-Bot-Api-Secret-Token": _WEBHOOK_SECRET}
_VALID_PAYLOAD = {
"email": "user@example.com",
"email": "user@tutlot.com",
"login": "testuser",
"password": "strongpassword123",
}
@ -67,7 +68,7 @@ async def test_auth_register_fire_and_forget_telegram_error_still_returns_201():
):
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "other@example.com", "login": "otheruser"},
json={**_VALID_PAYLOAD, "email": "other@tutlot.com", "login": "otheruser"},
)
await asyncio.sleep(0)
@ -105,7 +106,7 @@ async def test_auth_register_409_on_duplicate_login():
r2 = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "different@example.com"},
json={**_VALID_PAYLOAD, "email": "different@tutlot.com"},
)
assert r2.status_code == 409, f"Expected 409 on duplicate login, got {r2.status_code}"
@ -167,20 +168,23 @@ async def test_auth_register_422_short_password():
@pytest.mark.asyncio
async def test_auth_register_sends_notification_to_admin():
"""Registration triggers send_registration_notification with correct data."""
calls: list[dict] = []
"""Registration triggers real HTTP sendMessage to ADMIN_CHAT_ID with correct login/email."""
from backend import config as _cfg
async def _capture(reg_id, login, email, created_at):
calls.append({"reg_id": reg_id, "login": login, "email": email})
captured: list[dict] = []
async with make_app_client(capture_send_requests=captured) as client:
resp = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
assert resp.status_code == 201
await asyncio.sleep(0)
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", side_effect=_capture):
await client.post("/api/auth/register", json=_VALID_PAYLOAD)
await asyncio.sleep(0)
assert len(calls) == 1, f"Expected 1 notification call, got {len(calls)}"
assert calls[0]["login"] == _VALID_PAYLOAD["login"]
assert calls[0]["email"] == _VALID_PAYLOAD["email"]
admin_chat_id = str(_cfg.ADMIN_CHAT_ID)
admin_msgs = [r for r in captured if str(r.get("chat_id")) == admin_chat_id]
assert len(admin_msgs) >= 1, (
f"Expected sendMessage to ADMIN_CHAT_ID={admin_chat_id!r}, captured: {captured}"
)
text = admin_msgs[0].get("text", "")
assert _VALID_PAYLOAD["login"] in text, f"Expected login in text: {text!r}"
assert _VALID_PAYLOAD["email"] in text, f"Expected email in text: {text!r}"
# ---------------------------------------------------------------------------
@ -361,7 +365,7 @@ async def test_register_without_push_subscription():
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "nopush@example.com", "login": "nopushuser"},
json={**_VALID_PAYLOAD, "email": "nopush@tutlot.com", "login": "nopushuser"},
)
assert resp.status_code == 201
assert resp.json()["status"] == "pending"
@ -420,7 +424,7 @@ async def test_webhook_callback_approve_edits_message():
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "edit@example.com", "login": "edituser"},
json={**_VALID_PAYLOAD, "email": "edit@tutlot.com", "login": "edituser"},
)
assert reg_resp.status_code == 201
@ -465,7 +469,7 @@ async def test_webhook_callback_answer_sent():
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "answer@example.com", "login": "answeruser"},
json={**_VALID_PAYLOAD, "email": "answer@tutlot.com", "login": "answeruser"},
)
assert reg_resp.status_code == 201
@ -558,7 +562,7 @@ async def test_password_hash_stored_in_pbkdf2_format():
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "pbkdf2@example.com", "login": "pbkdf2user"},
json={**_VALID_PAYLOAD, "email": "pbkdf2@tutlot.com", "login": "pbkdf2user"},
)
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
@ -579,3 +583,306 @@ async def test_password_hash_stored_in_pbkdf2_format():
assert len(dk_hex) == 64, f"Expected 64-char dk hex (SHA-256), got {len(dk_hex)}"
int(salt_hex, 16) # raises ValueError if not valid hex
int(dk_hex, 16)
# ---------------------------------------------------------------------------
# 15. State machine — повторное нажатие approve на уже approved
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_webhook_callback_double_approve_does_not_send_push():
"""Second approve on already-approved registration must NOT fire push."""
push_sub = {
"endpoint": "https://fcm.googleapis.com/fcm/send/test2",
"keys": {"p256dh": "BQDEF", "auth": "abc"},
}
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "double@tutlot.com", "login": "doubleuser", "push_subscription": push_sub},
)
assert reg_resp.status_code == 201
from backend import config as _cfg
import aiosqlite
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
conn.row_factory = aiosqlite.Row
async with conn.execute("SELECT id FROM registrations LIMIT 1") as cur:
row = await cur.fetchone()
reg_id = row["id"] if row else None
assert reg_id is not None
cb_payload = {
"callback_query": {
"id": "cq_d1",
"data": f"approve:{reg_id}",
"message": {"message_id": 60, "chat": {"id": 5694335584}},
}
}
# First approve — should succeed
await client.post("/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS)
await asyncio.sleep(0)
# Second approve — push must NOT be fired
push_calls: list = []
async def _capture_push(sub_json, title, body):
push_calls.append(sub_json)
cb_payload2 = {**cb_payload, "callback_query": {**cb_payload["callback_query"], "id": "cq_d2"}}
with patch("backend.push.send_push", side_effect=_capture_push):
await client.post("/api/webhook/telegram", json=cb_payload2, headers=_WEBHOOK_HEADERS)
await asyncio.sleep(0)
assert len(push_calls) == 0, f"Second approve must not fire push, got {len(push_calls)} calls"
# Also verify status is still 'approved'
from backend import db as _db
# Can't check here as client context is closed; DB assertion was covered by state machine logic
@pytest.mark.asyncio
async def test_webhook_callback_double_approve_status_stays_approved():
"""Status remains 'approved' after a second approve callback."""
from backend import db as _db
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "stay@tutlot.com", "login": "stayuser"},
)
assert reg_resp.status_code == 201
from backend import config as _cfg
import aiosqlite
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
conn.row_factory = aiosqlite.Row
async with conn.execute("SELECT id FROM registrations LIMIT 1") as cur:
row = await cur.fetchone()
reg_id = row["id"] if row else None
assert reg_id is not None
cb = {
"callback_query": {
"id": "cq_s1",
"data": f"approve:{reg_id}",
"message": {"message_id": 70, "chat": {"id": 5694335584}},
}
}
await client.post("/api/webhook/telegram", json=cb, headers=_WEBHOOK_HEADERS)
await asyncio.sleep(0)
cb2 = {**cb, "callback_query": {**cb["callback_query"], "id": "cq_s2"}}
await client.post("/api/webhook/telegram", json=cb2, headers=_WEBHOOK_HEADERS)
await asyncio.sleep(0)
reg = await _db.get_registration(reg_id)
assert reg["status"] == "approved", f"Expected 'approved', got {reg['status']!r}"
# ---------------------------------------------------------------------------
# 16. State machine — approve после reject
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_webhook_callback_approve_after_reject_status_stays_rejected():
"""Approve after reject must NOT change status — remains 'rejected'."""
from backend import db as _db
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
reg_resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "artest@tutlot.com", "login": "artestuser"},
)
assert reg_resp.status_code == 201
from backend import config as _cfg
import aiosqlite
async with aiosqlite.connect(_cfg.DB_PATH) as conn:
conn.row_factory = aiosqlite.Row
async with conn.execute("SELECT id FROM registrations LIMIT 1") as cur:
row = await cur.fetchone()
reg_id = row["id"] if row else None
assert reg_id is not None
# First: reject
rej_cb = {
"callback_query": {
"id": "cq_ar1",
"data": f"reject:{reg_id}",
"message": {"message_id": 80, "chat": {"id": 5694335584}},
}
}
await client.post("/api/webhook/telegram", json=rej_cb, headers=_WEBHOOK_HEADERS)
await asyncio.sleep(0)
# Then: approve — must be ignored
push_calls: list = []
async def _capture_push(sub_json, title, body):
push_calls.append(sub_json)
app_cb = {
"callback_query": {
"id": "cq_ar2",
"data": f"approve:{reg_id}",
"message": {"message_id": 81, "chat": {"id": 5694335584}},
}
}
with patch("backend.push.send_push", side_effect=_capture_push):
await client.post("/api/webhook/telegram", json=app_cb, headers=_WEBHOOK_HEADERS)
await asyncio.sleep(0)
reg = await _db.get_registration(reg_id)
assert reg["status"] == "rejected", f"Expected 'rejected', got {reg['status']!r}"
assert len(push_calls) == 0, f"Approve after reject must not fire push, got {len(push_calls)}"
# ---------------------------------------------------------------------------
# 17. Rate limiting — 4th request returns 429
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_auth_register_rate_limit_fourth_request_returns_429():
"""4th registration request from same IP within the window returns 429."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
for i in range(3):
r = await client.post(
"/api/auth/register",
json={
"email": f"ratetest{i}@tutlot.com",
"login": f"ratetest{i}",
"password": "strongpassword123",
},
)
assert r.status_code == 201, f"Request {i+1} should succeed, got {r.status_code}"
# 4th request — must be rate-limited
r4 = await client.post(
"/api/auth/register",
json={
"email": "ratetest4@tutlot.com",
"login": "ratetest4",
"password": "strongpassword123",
},
)
assert r4.status_code == 429, f"Expected 429 on 4th request, got {r4.status_code}"
# ---------------------------------------------------------------------------
# 18. VAPID public key endpoint /api/push/public-key
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_vapid_public_key_new_endpoint_returns_200():
"""GET /api/push/public-key returns 200 with vapid_public_key field."""
async with make_app_client() as client:
resp = await client.get("/api/push/public-key")
assert resp.status_code == 200, f"Expected 200, got {resp.status_code}: {resp.text}"
body = resp.json()
assert "vapid_public_key" in body, f"Expected 'vapid_public_key' in response, got {body}"
# ---------------------------------------------------------------------------
# 19. Password max length — 129 chars → 422
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_auth_register_422_password_too_long():
"""Password of 129 characters returns 422."""
async with make_app_client() as client:
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "password": "a" * 129},
)
assert resp.status_code == 422, f"Expected 422 on 129-char password, got {resp.status_code}"
# ---------------------------------------------------------------------------
# 20. Login max length — 31 chars → 422
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_auth_register_422_login_too_long():
"""Login of 31 characters returns 422."""
async with make_app_client() as client:
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "login": "a" * 31},
)
assert resp.status_code == 422, f"Expected 422 on 31-char login, got {resp.status_code}"
# ---------------------------------------------------------------------------
# 21. Empty body — POST /api/auth/register with {} → 422
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_auth_register_422_empty_body():
"""Empty JSON body returns 422."""
async with make_app_client() as client:
resp = await client.post("/api/auth/register", json={})
assert resp.status_code == 422, f"Expected 422 on empty body, got {resp.status_code}"
# ---------------------------------------------------------------------------
# 22. Malformed callback_data — no colon → ok:True without crash
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_webhook_callback_malformed_data_no_colon_returns_ok():
"""callback_query with data='garbage' (no colon) returns ok:True gracefully."""
async with make_app_client() as client:
cb_payload = {
"callback_query": {
"id": "cq_mal1",
"data": "garbage",
"message": {"message_id": 90, "chat": {"id": 5694335584}},
}
}
resp = await client.post(
"/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS
)
assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
assert resp.json() == {"ok": True}, f"Expected ok:True, got {resp.json()}"
# ---------------------------------------------------------------------------
# 23. Non-numeric reg_id — data='approve:abc' → ok:True without crash
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_webhook_callback_non_numeric_reg_id_returns_ok():
"""callback_query with data='approve:abc' (non-numeric reg_id) returns ok:True."""
async with make_app_client() as client:
cb_payload = {
"callback_query": {
"id": "cq_nan1",
"data": "approve:abc",
"message": {"message_id": 91, "chat": {"id": 5694335584}},
}
}
resp = await client.post(
"/api/webhook/telegram", json=cb_payload, headers=_WEBHOOK_HEADERS
)
assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
assert resp.json() == {"ok": True}, f"Expected ok:True, got {resp.json()}"

View file

@ -0,0 +1,439 @@
"""
Tests for BATON-008: Frontend registration module.
Acceptance criteria:
1. index.html форма регистрации с полями email, login, password присутствует
2. index.html НЕТ захардкоженных VAPID-ключей в HTML-атрибутах (decision #1333)
3. app.js вызов /api/push/public-key (не старый /api/vapid-public-key) (decision #1331)
4. app.js guard для PushManager (decision #1332)
5. app.js обработчик для кнопки регистрации (#btn-register → _handleSignUp)
6. app.js переключение между view-login и view-register
7. app.js показ ошибок пользователю (_setRegStatus)
8. GET /api/push/public-key 200 с vapid_public_key (API контракт)
9. POST /api/auth/register с валидными данными 201 (API контракт)
10. POST /api/auth/register с дублирующим email 409
11. POST /api/auth/register с дублирующим login 409
12. POST /api/auth/register с невалидным email 422
"""
from __future__ import annotations
import re
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
PROJECT_ROOT = Path(__file__).parent.parent
INDEX_HTML = PROJECT_ROOT / "frontend" / "index.html"
APP_JS = PROJECT_ROOT / "frontend" / "app.js"
from tests.conftest import make_app_client
_VALID_PAYLOAD = {
"email": "frontend_test@tutlot.com",
"login": "frontenduser",
"password": "strongpassword123",
}
# ---------------------------------------------------------------------------
# HTML static analysis — Criterion 1: поля формы регистрации
# ---------------------------------------------------------------------------
def test_index_html_has_email_field() -> None:
"""index.html должен содержать поле email для регистрации (id=reg-email)."""
content = INDEX_HTML.read_text(encoding="utf-8")
assert 'id="reg-email"' in content, (
"index.html не содержит поле с id='reg-email'"
)
def test_index_html_has_login_field() -> None:
"""index.html должен содержать поле логина для регистрации (id=reg-login)."""
content = INDEX_HTML.read_text(encoding="utf-8")
assert 'id="reg-login"' in content, (
"index.html не содержит поле с id='reg-login'"
)
def test_index_html_has_password_field() -> None:
"""index.html должен содержать поле пароля для регистрации (id=reg-password)."""
content = INDEX_HTML.read_text(encoding="utf-8")
assert 'id="reg-password"' in content, (
"index.html не содержит поле с id='reg-password'"
)
def test_index_html_email_field_has_correct_type() -> None:
"""Поле email регистрации должно иметь type='email'."""
content = INDEX_HTML.read_text(encoding="utf-8")
# Ищем input с id=reg-email и type=email в любом порядке атрибутов
email_input_block = re.search(
r'<input[^>]*id="reg-email"[^>]*>', content, re.DOTALL
)
assert email_input_block is not None, "Не найден input с id='reg-email'"
assert 'type="email"' in email_input_block.group(0), (
"Поле reg-email не имеет type='email'"
)
def test_index_html_password_field_has_correct_type() -> None:
"""Поле пароля регистрации должно иметь type='password'."""
content = INDEX_HTML.read_text(encoding="utf-8")
password_input_block = re.search(
r'<input[^>]*id="reg-password"[^>]*>', content, re.DOTALL
)
assert password_input_block is not None, "Не найден input с id='reg-password'"
assert 'type="password"' in password_input_block.group(0), (
"Поле reg-password не имеет type='password'"
)
def test_index_html_has_register_button() -> None:
"""index.html должен содержать кнопку регистрации (id=btn-register)."""
content = INDEX_HTML.read_text(encoding="utf-8")
assert 'id="btn-register"' in content, (
"index.html не содержит кнопку с id='btn-register'"
)
def test_index_html_has_switch_to_register_button() -> None:
"""index.html должен содержать кнопку переключения на форму регистрации (id=btn-switch-to-register)."""
content = INDEX_HTML.read_text(encoding="utf-8")
assert 'id="btn-switch-to-register"' in content, (
"index.html не содержит кнопку с id='btn-switch-to-register'"
)
def test_index_html_has_view_register_div() -> None:
"""index.html должен содержать блок view-register для формы регистрации."""
content = INDEX_HTML.read_text(encoding="utf-8")
assert 'id="view-register"' in content, (
"index.html не содержит блок с id='view-register'"
)
def test_index_html_has_view_login_div() -> None:
"""index.html должен содержать блок view-login для онбординга."""
content = INDEX_HTML.read_text(encoding="utf-8")
assert 'id="view-login"' in content, (
"index.html не содержит блок с id='view-login'"
)
def test_index_html_has_reg_status_element() -> None:
"""index.html должен содержать элемент статуса регистрации (id=reg-status)."""
content = INDEX_HTML.read_text(encoding="utf-8")
assert 'id="reg-status"' in content, (
"index.html не содержит элемент с id='reg-status'"
)
# ---------------------------------------------------------------------------
# HTML static analysis — Criterion 2: НЕТ захардкоженного VAPID в HTML (decision #1333)
# ---------------------------------------------------------------------------
def test_index_html_no_hardcoded_vapid_key_in_meta() -> None:
"""index.html НЕ должен содержать VAPID-ключ захардкоженным в meta-теге (decision #1333)."""
content = INDEX_HTML.read_text(encoding="utf-8")
# VAPID public key — это URL-safe base64 строка длиной 87 символов (без padding)
# Ищем характерный паттерн в meta-атрибутах
vapid_in_meta = re.search(
r'<meta[^>]+content\s*=\s*["\'][A-Za-z0-9_\-]{60,}["\'][^>]*>',
content,
)
assert vapid_in_meta is None, (
f"Найден meta-тег с длинной строкой (возможный VAPID-ключ): "
f"{vapid_in_meta.group(0) if vapid_in_meta else ''}"
)
def test_index_html_no_vapid_key_attribute_pattern() -> None:
"""index.html НЕ должен содержать data-vapid-key или аналогичные атрибуты."""
content = INDEX_HTML.read_text(encoding="utf-8")
assert "vapid" not in content.lower(), (
"index.html содержит упоминание 'vapid' — VAPID ключ должен читаться через API, "
"а не быть захардкожен в HTML (decision #1333)"
)
# ---------------------------------------------------------------------------
# app.js static analysis — Criterion 3: /api/push/public-key endpoint (decision #1331)
# ---------------------------------------------------------------------------
def test_app_js_uses_new_vapid_endpoint() -> None:
"""app.js должен обращаться к /api/push/public-key (decision #1331)."""
content = APP_JS.read_text(encoding="utf-8")
assert "/api/push/public-key" in content, (
"app.js не содержит endpoint '/api/push/public-key'"
)
def test_app_js_does_not_use_old_vapid_endpoint() -> None:
"""app.js НЕ должен использовать устаревший /api/vapid-public-key (decision #1331)."""
content = APP_JS.read_text(encoding="utf-8")
assert "/api/vapid-public-key" not in content, (
"app.js содержит устаревший endpoint '/api/vapid-public-key'"
"нарушение decision #1331, должен использоваться '/api/push/public-key'"
)
# ---------------------------------------------------------------------------
# app.js static analysis — Criterion 4: PushManager guard (decision #1332)
# ---------------------------------------------------------------------------
def test_app_js_has_push_manager_guard_in_registration_flow() -> None:
"""app.js должен содержать guard 'PushManager' in window (decision #1332)."""
content = APP_JS.read_text(encoding="utf-8")
assert "'PushManager' in window" in content, (
"app.js не содержит guard \"'PushManager' in window\""
"нарушение decision #1332"
)
def test_app_js_push_manager_guard_combined_with_service_worker_check() -> None:
"""Guard PushManager должен сочетаться с проверкой serviceWorker."""
content = APP_JS.read_text(encoding="utf-8")
# Ищем паттерн совместной проверки serviceWorker + PushManager
assert re.search(
r"serviceWorker.*PushManager|PushManager.*serviceWorker",
content,
re.DOTALL,
), (
"app.js не содержит совместной проверки 'serviceWorker' и 'PushManager'"
"guard неполный (decision #1332)"
)
# ---------------------------------------------------------------------------
# app.js static analysis — Criterion 5: обработчик кнопки регистрации
# ---------------------------------------------------------------------------
def test_app_js_has_handle_sign_up_function() -> None:
"""app.js должен содержать функцию _handleSignUp."""
content = APP_JS.read_text(encoding="utf-8")
assert "_handleSignUp" in content, (
"app.js не содержит функцию '_handleSignUp'"
)
def test_app_js_registers_click_handler_for_btn_register() -> None:
"""app.js должен добавлять click-обработчик на btn-register → _handleSignUp."""
content = APP_JS.read_text(encoding="utf-8")
# Ищем addEventListener на элементе btn-register с вызовом _handleSignUp
assert re.search(
r'btn-register.*addEventListener|addEventListener.*btn-register',
content,
re.DOTALL,
), (
"app.js не содержит addEventListener для кнопки 'btn-register'"
)
# Проверяем что именно _handleSignUp привязан к кнопке
assert re.search(
r'btn[Rr]egister.*_handleSignUp|_handleSignUp.*btn[Rr]egister',
content,
re.DOTALL,
), (
"app.js не связывает кнопку 'btn-register' с функцией '_handleSignUp'"
)
# ---------------------------------------------------------------------------
# app.js static analysis — Criterion 6: переключение view-login / view-register
# ---------------------------------------------------------------------------
def test_app_js_has_show_view_function() -> None:
"""app.js должен содержать функцию _showView для переключения видов."""
content = APP_JS.read_text(encoding="utf-8")
assert "_showView" in content, (
"app.js не содержит функцию '_showView'"
)
def test_app_js_show_view_handles_view_login() -> None:
"""_showView в app.js должна обрабатывать view-login."""
content = APP_JS.read_text(encoding="utf-8")
assert "view-login" in content, (
"app.js не содержит id 'view-login' — нет переключения на вид логина"
)
def test_app_js_show_view_handles_view_register() -> None:
"""_showView в app.js должна обрабатывать view-register."""
content = APP_JS.read_text(encoding="utf-8")
assert "view-register" in content, (
"app.js не содержит id 'view-register' — нет переключения на вид регистрации"
)
def test_app_js_has_btn_switch_to_register_handler() -> None:
"""app.js должен содержать обработчик для btn-switch-to-register."""
content = APP_JS.read_text(encoding="utf-8")
assert "btn-switch-to-register" in content, (
"app.js не содержит ссылку на 'btn-switch-to-register'"
)
def test_app_js_has_btn_switch_to_login_handler() -> None:
"""app.js должен содержать обработчик для btn-switch-to-login (назад)."""
content = APP_JS.read_text(encoding="utf-8")
assert "btn-switch-to-login" in content, (
"app.js не содержит ссылку на 'btn-switch-to-login'"
)
# ---------------------------------------------------------------------------
# app.js static analysis — Criterion 7: обработка ошибок / показ сообщения пользователю
# ---------------------------------------------------------------------------
def test_app_js_has_set_reg_status_function() -> None:
"""app.js должен содержать _setRegStatus для показа статуса в форме регистрации."""
content = APP_JS.read_text(encoding="utf-8")
assert "_setRegStatus" in content, (
"app.js не содержит функцию '_setRegStatus'"
)
def test_app_js_handle_sign_up_shows_error_on_empty_fields() -> None:
"""_handleSignUp должна вызывать _setRegStatus с ошибкой при пустых полях."""
content = APP_JS.read_text(encoding="utf-8")
# Проверяем наличие валидации пустых полей внутри _handleSignUp-подобного блока
assert re.search(
r"_setRegStatus\s*\([^)]*error",
content,
), (
"app.js не содержит вызов _setRegStatus с классом 'error' "
"— ошибки не отображаются пользователю"
)
def test_app_js_handle_sign_up_shows_success_on_ok() -> None:
"""_handleSignUp должна вызывать _setRegStatus с success при успешной регистрации."""
content = APP_JS.read_text(encoding="utf-8")
assert re.search(
r"_setRegStatus\s*\([^)]*success",
content,
), (
"app.js не содержит вызов _setRegStatus с классом 'success' "
"— пользователь не уведомляется об успехе регистрации"
)
def test_app_js_clears_password_after_successful_signup() -> None:
"""_handleSignUp должна очищать поле пароля после успешной отправки."""
content = APP_JS.read_text(encoding="utf-8")
# Ищем сброс значения пароля
assert re.search(
r"passwordInput\.value\s*=\s*['\"][\s]*['\"]",
content,
), (
"app.js не очищает поле пароля после успешной регистрации — "
"пароль остаётся в DOM (security concern)"
)
def test_app_js_uses_api_auth_register_endpoint() -> None:
"""app.js должен отправлять форму на /api/auth/register."""
content = APP_JS.read_text(encoding="utf-8")
assert "/api/auth/register" in content, (
"app.js не содержит endpoint '/api/auth/register'"
)
# ---------------------------------------------------------------------------
# Integration tests — API контракты (Criteria 812)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_vapid_public_key_endpoint_returns_200_with_key():
"""GET /api/push/public-key → 200 с полем vapid_public_key."""
async with make_app_client() as client:
resp = await client.get("/api/push/public-key")
assert resp.status_code == 200, (
f"GET /api/push/public-key вернул {resp.status_code}, ожидался 200"
)
body = resp.json()
assert "vapid_public_key" in body, (
f"Ответ /api/push/public-key не содержит 'vapid_public_key': {body}"
)
assert isinstance(body["vapid_public_key"], str), (
"vapid_public_key должен быть строкой"
)
@pytest.mark.asyncio
async def test_register_valid_payload_returns_201_pending():
"""POST /api/auth/register с валидными данными → 201 status=pending."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
resp = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
assert resp.status_code == 201, (
f"POST /api/auth/register вернул {resp.status_code}: {resp.text}"
)
body = resp.json()
assert body.get("status") == "pending", (
f"Ожидался status='pending', получено: {body}"
)
assert "message" in body, (
f"Ответ не содержит поле 'message': {body}"
)
@pytest.mark.asyncio
async def test_register_duplicate_email_returns_409():
"""POST /api/auth/register с дублирующим email → 409."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
r1 = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
assert r1.status_code == 201, f"Первая регистрация не прошла: {r1.text}"
r2 = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "login": "anotherlogin"},
)
assert r2.status_code == 409, (
f"Дублирующий email должен вернуть 409, получено {r2.status_code}"
)
@pytest.mark.asyncio
async def test_register_duplicate_login_returns_409():
"""POST /api/auth/register с дублирующим login → 409."""
async with make_app_client() as client:
with patch("backend.telegram.send_registration_notification", new_callable=AsyncMock):
r1 = await client.post("/api/auth/register", json=_VALID_PAYLOAD)
assert r1.status_code == 201, f"Первая регистрация не прошла: {r1.text}"
r2 = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "another@tutlot.com"},
)
assert r2.status_code == 409, (
f"Дублирующий login должен вернуть 409, получено {r2.status_code}"
)
@pytest.mark.asyncio
async def test_register_invalid_email_returns_422():
"""POST /api/auth/register с невалидным email → 422."""
async with make_app_client() as client:
resp = await client.post(
"/api/auth/register",
json={**_VALID_PAYLOAD, "email": "not-an-email"},
)
assert resp.status_code == 422, (
f"Невалидный email должен вернуть 422, получено {resp.status_code}"
)

338
tests/test_biz_001.py Normal file
View file

@ -0,0 +1,338 @@
"""
Tests for BATON-BIZ-001: Login mechanism for approved users (dual-layer: AST + httpx functional).
Acceptance criteria:
1. Успешный login по login-полю 200 + token
2. Успешный login по email-полю 200 + token
3. Неверный пароль 401 (без раскрытия причины)
4. Статус pending 403 с читаемым сообщением
5. Статус rejected 403 с читаемым сообщением
6. Rate limit 6-й запрос подряд 429
7. Guard middleware возвращает 401 без токена
8. Guard middleware пропускает валидный токен
Additional: error message uniformity, PBKDF2 verification.
"""
from __future__ import annotations
import os
os.environ.setdefault("BOT_TOKEN", "test-bot-token")
os.environ.setdefault("CHAT_ID", "-1001234567890")
os.environ.setdefault("WEBHOOK_SECRET", "test-webhook-secret")
os.environ.setdefault("WEBHOOK_URL", "https://example.com/api/webhook/telegram")
os.environ.setdefault("FRONTEND_ORIGIN", "http://localhost:3000")
os.environ.setdefault("ADMIN_TOKEN", "test-admin-token")
os.environ.setdefault("ADMIN_CHAT_ID", "5694335584")
import pytest
from fastapi import HTTPException
from fastapi.security import HTTPAuthorizationCredentials
from backend import db
from backend.middleware import create_auth_token, verify_auth_token
from tests.conftest import make_app_client
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
async def _register_auth(client, email: str, login: str, password: str) -> int:
"""Register via /api/auth/register, return registration id."""
resp = await client.post(
"/api/auth/register",
json={"email": email, "login": login, "password": password},
)
assert resp.status_code == 201, f"auth/register failed: {resp.text}"
reg = await db.get_registration_by_login_or_email(login)
assert reg is not None
return reg["id"]
async def _approve(reg_id: int) -> None:
await db.update_registration_status(reg_id, "approved")
async def _reject(reg_id: int) -> None:
await db.update_registration_status(reg_id, "rejected")
# ---------------------------------------------------------------------------
# Criterion 1 — Успешный login по login-полю
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_login_by_login_field_returns_200_with_token():
"""Approved user can login using their login field → 200 + token."""
async with make_app_client() as client:
reg_id = await _register_auth(client, "alice@tutlot.com", "alice", "password123")
await _approve(reg_id)
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "alice", "password": "password123"},
)
assert resp.status_code == 200
data = resp.json()
assert "token" in data
assert data["login"] == "alice"
@pytest.mark.asyncio
async def test_login_by_login_field_token_is_non_empty_string():
"""Token returned for approved login user is a non-empty string."""
async with make_app_client() as client:
reg_id = await _register_auth(client, "alice2@tutlot.com", "alice2", "password123")
await _approve(reg_id)
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "alice2", "password": "password123"},
)
assert isinstance(resp.json()["token"], str)
assert len(resp.json()["token"]) > 0
# ---------------------------------------------------------------------------
# Criterion 2 — Успешный login по email-полю
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_login_by_email_field_returns_200_with_token():
"""Approved user can login using their email field → 200 + token."""
async with make_app_client() as client:
reg_id = await _register_auth(client, "bob@tutlot.com", "bobuser", "securepass1")
await _approve(reg_id)
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "bob@tutlot.com", "password": "securepass1"},
)
assert resp.status_code == 200
data = resp.json()
assert "token" in data
assert data["login"] == "bobuser"
@pytest.mark.asyncio
async def test_login_by_email_field_token_login_matches_registration():
"""Token response login field matches the login set during registration."""
async with make_app_client() as client:
reg_id = await _register_auth(client, "bob2@tutlot.com", "bob2user", "securepass1")
await _approve(reg_id)
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "bob2@tutlot.com", "password": "securepass1"},
)
assert resp.json()["login"] == "bob2user"
# ---------------------------------------------------------------------------
# Criterion 3 — Неверный пароль → 401 без раскрытия причины
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_wrong_password_returns_401():
"""Wrong password returns 401 with generic message (no detail about which field failed)."""
async with make_app_client() as client:
reg_id = await _register_auth(client, "carol@tutlot.com", "carol", "correctpass1")
await _approve(reg_id)
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "carol", "password": "wrongpassword"},
)
assert resp.status_code == 401
assert "Неверный логин или пароль" in resp.json()["detail"]
@pytest.mark.asyncio
async def test_nonexistent_user_returns_401_same_message_as_wrong_password():
"""Non-existent login returns same 401 message as wrong password (prevents user enumeration)."""
async with make_app_client() as client:
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "doesnotexist", "password": "anypassword"},
)
assert resp.status_code == 401
assert "Неверный логин или пароль" in resp.json()["detail"]
# ---------------------------------------------------------------------------
# Criterion 4 — Статус pending → 403 с читаемым сообщением
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_pending_user_login_returns_403():
"""User with pending status gets 403."""
async with make_app_client() as client:
await _register_auth(client, "dave@tutlot.com", "dave", "password123")
# Status is 'pending' by default — no approval step
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "dave", "password": "password123"},
)
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_pending_user_login_403_message_is_human_readable():
"""403 message for pending user contains readable Russian text about the waiting status."""
async with make_app_client() as client:
await _register_auth(client, "dave2@tutlot.com", "dave2", "password123")
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "dave2", "password": "password123"},
)
assert "ожидает" in resp.json()["detail"]
# ---------------------------------------------------------------------------
# Criterion 5 — Статус rejected → 403 с читаемым сообщением
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_rejected_user_login_returns_403():
"""User with rejected status gets 403."""
async with make_app_client() as client:
reg_id = await _register_auth(client, "eve@tutlot.com", "evegirl", "password123")
await _reject(reg_id)
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "evegirl", "password": "password123"},
)
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_rejected_user_login_403_message_is_human_readable():
"""403 message for rejected user contains readable Russian text about rejection."""
async with make_app_client() as client:
reg_id = await _register_auth(client, "eve2@tutlot.com", "eve2girl", "password123")
await _reject(reg_id)
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "eve2girl", "password": "password123"},
)
assert "отклонена" in resp.json()["detail"]
# ---------------------------------------------------------------------------
# Criterion 6 — Rate limit: 6-й запрос подряд → 429
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_rate_limit_triggers_on_sixth_login_attempt():
"""Login rate limit (5 per window) triggers 429 exactly on the 6th request."""
async with make_app_client() as client:
statuses = []
for _ in range(6):
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "nouser_rl", "password": "nopass"},
headers={"X-Real-IP": "10.99.99.1"},
)
statuses.append(resp.status_code)
# First 5 attempts pass rate limit (user not found → 401)
assert all(s == 401 for s in statuses[:5]), (
f"Первые 5 попыток должны быть 401, получили: {statuses[:5]}"
)
# 6th attempt hits rate limit
assert statuses[5] == 429, (
f"6-я попытка должна быть 429, получили: {statuses[5]}"
)
@pytest.mark.asyncio
async def test_rate_limit_fifth_attempt_still_passes():
"""5th login attempt is still allowed (rate limit triggers only on 6th)."""
async with make_app_client() as client:
for i in range(4):
await client.post(
"/api/auth/login",
json={"login_or_email": "nouser_rl2", "password": "nopass"},
headers={"X-Real-IP": "10.99.99.2"},
)
resp = await client.post(
"/api/auth/login",
json={"login_or_email": "nouser_rl2", "password": "nopass"},
headers={"X-Real-IP": "10.99.99.2"},
)
assert resp.status_code == 401, (
f"5-я попытка должна пройти rate limit и вернуть 401, получили: {resp.status_code}"
)
# ---------------------------------------------------------------------------
# Criterion 7 — Guard middleware: 401 без токена
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_verify_auth_token_raises_401_when_credentials_is_none():
"""verify_auth_token raises HTTPException 401 when no credentials provided."""
with pytest.raises(HTTPException) as exc_info:
await verify_auth_token(credentials=None)
assert exc_info.value.status_code == 401
@pytest.mark.asyncio
async def test_verify_auth_token_raises_401_for_malformed_token():
"""verify_auth_token raises HTTPException 401 for a malformed/invalid token."""
creds = HTTPAuthorizationCredentials(scheme="Bearer", credentials="not.a.valid.jwt")
with pytest.raises(HTTPException) as exc_info:
await verify_auth_token(credentials=creds)
assert exc_info.value.status_code == 401
# ---------------------------------------------------------------------------
# Criterion 8 — Guard middleware: валидный токен пропускается
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_verify_auth_token_returns_payload_for_valid_token():
"""verify_auth_token returns decoded JWT payload for a valid signed token."""
token = create_auth_token(reg_id=42, login="testuser")
creds = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
payload = await verify_auth_token(credentials=creds)
assert payload["sub"] == "42"
assert payload["login"] == "testuser"
@pytest.mark.asyncio
async def test_verify_auth_token_payload_contains_expected_fields():
"""Payload returned by verify_auth_token contains sub, login, iat, exp fields."""
token = create_auth_token(reg_id=7, login="inspector")
creds = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
payload = await verify_auth_token(credentials=creds)
for field in ("sub", "login", "iat", "exp"):
assert field in payload, f"Поле '{field}' отсутствует в payload"
# ---------------------------------------------------------------------------
# Additional: PBKDF2 correctness — verify_password timing-safe
# ---------------------------------------------------------------------------
def test_hash_and_verify_password_returns_true_for_correct_password():
"""_hash_password + _verify_password: correct password returns True."""
from backend.main import _hash_password, _verify_password
stored = _hash_password("mysecretpass")
assert _verify_password("mysecretpass", stored) is True
def test_hash_and_verify_password_returns_false_for_wrong_password():
"""_hash_password + _verify_password: wrong password returns False."""
from backend.main import _hash_password, _verify_password
stored = _hash_password("mysecretpass")
assert _verify_password("wrongpassword", stored) is False
def test_verify_password_returns_false_for_malformed_hash():
"""_verify_password returns False (not exception) for a malformed hash string."""
from backend.main import _verify_password
assert _verify_password("anypassword", "not-a-valid-hash") is False

203
tests/test_biz_002.py Normal file
View file

@ -0,0 +1,203 @@
"""
Tests for BATON-BIZ-002: Убрать hardcoded VAPID key из meta-тега, читать с /api/push/public-key
Acceptance criteria:
1. Meta-тег vapid-public-key полностью отсутствует в frontend/index.html (decision #1333).
2. app.js использует canonical URL /api/push/public-key для получения VAPID ключа.
3. Graceful fallback: endpoint недоступен функция возвращает null, не бросает исключение.
4. Graceful fallback: ключ пустой _initPushSubscription не выполняется (guard на null).
5. GET /api/push/public-key возвращает HTTP 200 с полем vapid_public_key.
6. GET /api/push/public-key возвращает правильное значение из конфига.
"""
from __future__ import annotations
import os
import re
from pathlib import Path
from unittest.mock import patch
os.environ.setdefault("BOT_TOKEN", "test-bot-token")
os.environ.setdefault("CHAT_ID", "-1001234567890")
os.environ.setdefault("WEBHOOK_SECRET", "test-webhook-secret")
os.environ.setdefault("WEBHOOK_URL", "https://example.com/api/webhook/telegram")
os.environ.setdefault("FRONTEND_ORIGIN", "http://localhost:3000")
os.environ.setdefault("ADMIN_TOKEN", "test-admin-token")
os.environ.setdefault("ADMIN_CHAT_ID", "5694335584")
import pytest
from tests.conftest import make_app_client
PROJECT_ROOT = Path(__file__).parent.parent
INDEX_HTML = PROJECT_ROOT / "frontend" / "index.html"
APP_JS = PROJECT_ROOT / "frontend" / "app.js"
_TEST_VAPID_KEY = "BFakeVapidPublicKeyForBiz002TestingBase64UrlEncoded"
# ---------------------------------------------------------------------------
# Criterion 1 — AST: meta-тег vapid-public-key полностью отсутствует
# ---------------------------------------------------------------------------
def test_index_html_has_no_meta_tag_named_vapid_public_key() -> None:
"""index.html не должен содержать <meta name='vapid-public-key'> вообще (decision #1333)."""
content = INDEX_HTML.read_text(encoding="utf-8")
match = re.search(
r'<meta[^>]+name\s*=\s*["\']vapid-public-key["\']',
content,
re.IGNORECASE,
)
assert match is None, (
f"index.html содержит удалённый тег <meta name='vapid-public-key'>: {match.group(0)!r}"
)
def test_index_html_has_no_vapid_meta_tag_with_empty_or_any_content() -> None:
"""index.html не должен содержать ни пустой, ни непустой VAPID ключ в meta content."""
content = INDEX_HTML.read_text(encoding="utf-8")
match = re.search(
r'<meta[^>]*(?:vapid|application-server-key)[^>]*content\s*=',
content,
re.IGNORECASE,
)
assert match is None, (
f"index.html содержит <meta>-тег с VAPID-связанным атрибутом content: {match.group(0)!r}"
)
# ---------------------------------------------------------------------------
# Criterion 2 — AST: app.js использует canonical /api/push/public-key
# ---------------------------------------------------------------------------
def test_app_js_fetch_vapid_uses_canonical_push_public_key_url() -> None:
"""_fetchVapidPublicKey в app.js должна использовать /api/push/public-key (canonical URL)."""
content = APP_JS.read_text(encoding="utf-8")
assert "/api/push/public-key" in content, (
"app.js не содержит canonical URL '/api/push/public-key'"
"ключ не читается через правильный endpoint"
)
def test_app_js_fetch_vapid_returns_vapid_public_key_field() -> None:
"""_fetchVapidPublicKey должна читать поле vapid_public_key из JSON-ответа."""
content = APP_JS.read_text(encoding="utf-8")
assert re.search(r"data\.vapid_public_key", content), (
"app.js не читает поле 'data.vapid_public_key' из ответа API"
)
# ---------------------------------------------------------------------------
# Criterion 3 — AST: graceful fallback когда endpoint недоступен
# ---------------------------------------------------------------------------
def test_app_js_fetch_vapid_returns_null_on_http_error() -> None:
"""_fetchVapidPublicKey должна возвращать null при res.ok === false (HTTP-ошибка)."""
content = APP_JS.read_text(encoding="utf-8")
assert re.search(r"if\s*\(\s*!\s*res\.ok\s*\)", content), (
"app.js не содержит проверку 'if (!res.ok)'"
"HTTP-ошибки не обрабатываются gracefully в _fetchVapidPublicKey"
)
def test_app_js_fetch_vapid_catches_network_errors() -> None:
"""_fetchVapidPublicKey должна оборачивать fetch в try/catch и возвращать null при сетевой ошибке."""
content = APP_JS.read_text(encoding="utf-8")
# Проверяем паттерн try { fetch ... } catch (err) { return null; } внутри функции
func_match = re.search(
r"async function _fetchVapidPublicKey\(\).*?(?=^(?:async )?function |\Z)",
content,
re.DOTALL | re.MULTILINE,
)
assert func_match, "Функция _fetchVapidPublicKey не найдена в app.js"
func_body = func_match.group(0)
assert "catch" in func_body, (
"app.js: _fetchVapidPublicKey не содержит блок catch — "
"сетевые ошибки при fetch не обрабатываются"
)
assert re.search(r"return\s+null", func_body), (
"app.js: _fetchVapidPublicKey не возвращает null при ошибке — "
"upstream код получит исключение вместо null"
)
# ---------------------------------------------------------------------------
# Criterion 4 — AST: graceful fallback когда ключ пустой (decision #1332)
# ---------------------------------------------------------------------------
def test_app_js_fetch_vapid_returns_null_on_empty_key() -> None:
"""_fetchVapidPublicKey должна возвращать null когда vapid_public_key пустой."""
content = APP_JS.read_text(encoding="utf-8")
assert re.search(r"data\.vapid_public_key\s*\|\|\s*null", content), (
"app.js не содержит 'data.vapid_public_key || null'"
"пустой ключ не преобразуется в null"
)
def test_app_js_init_push_subscription_guard_skips_on_null_key() -> None:
"""_initPushSubscription должна ранним возвратом пропускать подписку при null ключе (decision #1332)."""
content = APP_JS.read_text(encoding="utf-8")
assert re.search(r"if\s*\(\s*!\s*vapidPublicKey\s*\)", content), (
"app.js: _initPushSubscription не содержит guard 'if (!vapidPublicKey)'"
"подписка может быть создана без ключа"
)
# ---------------------------------------------------------------------------
# Criterion 5 — HTTP: GET /api/push/public-key → 200 + vapid_public_key
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_push_public_key_endpoint_returns_200() -> None:
"""GET /api/push/public-key должен вернуть HTTP 200."""
async with make_app_client() as client:
response = await client.get("/api/push/public-key")
assert response.status_code == 200, (
f"GET /api/push/public-key вернул {response.status_code}, ожидался 200"
)
@pytest.mark.asyncio
async def test_push_public_key_endpoint_returns_json_with_vapid_field() -> None:
"""GET /api/push/public-key должен вернуть JSON с полем vapid_public_key."""
async with make_app_client() as client:
response = await client.get("/api/push/public-key")
data = response.json()
assert "vapid_public_key" in data, (
f"Ответ /api/push/public-key не содержит поле 'vapid_public_key': {data!r}"
)
# ---------------------------------------------------------------------------
# Criterion 6 — HTTP: возвращает правильное значение из конфига
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_push_public_key_endpoint_returns_configured_value() -> None:
"""GET /api/push/public-key возвращает значение из VAPID_PUBLIC_KEY конфига."""
with patch("backend.config.VAPID_PUBLIC_KEY", _TEST_VAPID_KEY):
async with make_app_client() as client:
response = await client.get("/api/push/public-key")
data = response.json()
assert data.get("vapid_public_key") == _TEST_VAPID_KEY, (
f"vapid_public_key должен быть '{_TEST_VAPID_KEY}', "
f"получили: {data.get('vapid_public_key')!r}"
)
@pytest.mark.asyncio
async def test_push_public_key_endpoint_returns_empty_string_when_not_configured() -> None:
"""GET /api/push/public-key возвращает пустую строку (не ошибку) если ключ не настроен."""
with patch("backend.config.VAPID_PUBLIC_KEY", ""):
async with make_app_client() as client:
response = await client.get("/api/push/public-key")
assert response.status_code == 200, (
f"Endpoint вернул {response.status_code} при пустом ключе, ожидался 200"
)
data = response.json()
assert "vapid_public_key" in data, "Поле vapid_public_key отсутствует при пустом конфиге"

96
tests/test_biz_004.py Normal file
View file

@ -0,0 +1,96 @@
"""
BATON-BIZ-004: Verify removal of dead code from backend/telegram.py.
Acceptance criteria:
1. telegram.py does NOT contain duplicate logging setLevel calls for httpx/httpcore.
2. telegram.py does NOT contain the SignalAggregator class.
3. httpx/httpcore logging suppression is still configured in main.py (globally).
4. SignalAggregator is NOT importable from backend.telegram.
"""
from __future__ import annotations
import ast
import importlib
import inspect
import logging
import os
from pathlib import Path
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_BACKEND_DIR = Path(__file__).parent.parent / "backend"
_TELEGRAM_SRC = (_BACKEND_DIR / "telegram.py").read_text(encoding="utf-8")
_MAIN_SRC = (_BACKEND_DIR / "main.py").read_text(encoding="utf-8")
# ---------------------------------------------------------------------------
# Criteria 1 — no setLevel for httpx/httpcore in telegram.py
# ---------------------------------------------------------------------------
def test_telegram_has_no_httpx_setlevel():
"""telegram.py must not set log level for 'httpx'."""
assert 'getLogger("httpx").setLevel' not in _TELEGRAM_SRC
assert "getLogger('httpx').setLevel" not in _TELEGRAM_SRC
def test_telegram_has_no_httpcore_setlevel():
"""telegram.py must not set log level for 'httpcore'."""
assert 'getLogger("httpcore").setLevel' not in _TELEGRAM_SRC
assert "getLogger('httpcore').setLevel" not in _TELEGRAM_SRC
# ---------------------------------------------------------------------------
# Criteria 2 — SignalAggregator absent from telegram.py source
# ---------------------------------------------------------------------------
def test_telegram_source_has_no_signal_aggregator_class():
"""telegram.py source text must not contain the class definition."""
assert "class SignalAggregator" not in _TELEGRAM_SRC
def test_telegram_source_has_no_signal_aggregator_reference():
"""telegram.py source text must not reference SignalAggregator at all."""
assert "SignalAggregator" not in _TELEGRAM_SRC
# ---------------------------------------------------------------------------
# Criteria 3 — httpx/httpcore suppression still lives in main.py
# ---------------------------------------------------------------------------
def test_main_suppresses_httpx_logging():
"""main.py must call getLogger('httpx').setLevel to suppress noise."""
assert (
'getLogger("httpx").setLevel' in _MAIN_SRC
or "getLogger('httpx').setLevel" in _MAIN_SRC
)
def test_main_suppresses_httpcore_logging():
"""main.py must call getLogger('httpcore').setLevel to suppress noise."""
assert (
'getLogger("httpcore").setLevel' in _MAIN_SRC
or "getLogger('httpcore').setLevel" in _MAIN_SRC
)
# ---------------------------------------------------------------------------
# Criteria 4 — SignalAggregator not importable from backend.telegram
# ---------------------------------------------------------------------------
def test_signal_aggregator_not_importable_from_telegram():
"""Importing SignalAggregator from backend.telegram must raise ImportError."""
import importlib
import sys
# Force a fresh import so changes to the module are reflected
mod_name = "backend.telegram"
if mod_name in sys.modules:
del sys.modules[mod_name]
import backend.telegram as tg_mod # noqa: F401
assert not hasattr(tg_mod, "SignalAggregator"), (
"SignalAggregator should not be an attribute of backend.telegram"
)

155
tests/test_fix_007.py Normal file
View file

@ -0,0 +1,155 @@
"""
Tests for BATON-FIX-007: CORS OPTIONS preflight verification.
Acceptance criteria:
1. OPTIONS preflight to /api/signal returns 200.
2. Preflight response includes Access-Control-Allow-Methods containing GET.
3. Preflight response includes Access-Control-Allow-Origin matching the configured origin.
4. Preflight response includes Access-Control-Allow-Headers with Authorization.
5. allow_methods in CORSMiddleware configuration explicitly contains GET.
"""
from __future__ import annotations
import ast
from pathlib import Path
import pytest
from tests.conftest import make_app_client
_FRONTEND_ORIGIN = "http://localhost:3000"
_BACKEND_DIR = Path(__file__).parent.parent / "backend"
# ---------------------------------------------------------------------------
# Static check — CORSMiddleware config contains GET in allow_methods
# ---------------------------------------------------------------------------
def test_main_py_cors_allow_methods_contains_get() -> None:
"""allow_methods в CORSMiddleware должен содержать 'GET'."""
source = (_BACKEND_DIR / "main.py").read_text(encoding="utf-8")
tree = ast.parse(source, filename="main.py")
for node in ast.walk(tree):
if isinstance(node, ast.Call):
func = node.func
if isinstance(func, ast.Name) and func.id == "add_middleware":
continue
if not (
isinstance(func, ast.Attribute) and func.attr == "add_middleware"
):
continue
for kw in node.keywords:
if kw.arg == "allow_methods":
if isinstance(kw.value, ast.List):
methods = [
elt.value
for elt in kw.value.elts
if isinstance(elt, ast.Constant) and isinstance(elt.value, str)
]
assert "GET" in methods, (
f"allow_methods в CORSMiddleware не содержит 'GET': {methods}"
)
return
pytest.fail("add_middleware с CORSMiddleware и allow_methods не найден в main.py")
def test_main_py_cors_allow_methods_contains_post() -> None:
"""allow_methods в CORSMiddleware должен содержать 'POST' (регрессия)."""
source = (_BACKEND_DIR / "main.py").read_text(encoding="utf-8")
assert '"POST"' in source or "'POST'" in source, (
"allow_methods в CORSMiddleware не содержит 'POST'"
)
# ---------------------------------------------------------------------------
# Functional — OPTIONS preflight request
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_options_preflight_signal_returns_200() -> None:
"""OPTIONS preflight к /api/signal должен возвращать 200."""
async with make_app_client() as client:
resp = await client.options(
"/api/signal",
headers={
"Origin": _FRONTEND_ORIGIN,
"Access-Control-Request-Method": "POST",
"Access-Control-Request-Headers": "Content-Type, Authorization",
},
)
assert resp.status_code == 200, (
f"Preflight OPTIONS /api/signal вернул {resp.status_code}, ожидался 200"
)
@pytest.mark.asyncio
async def test_options_preflight_allow_origin_header() -> None:
"""OPTIONS preflight должен вернуть Access-Control-Allow-Origin."""
async with make_app_client() as client:
resp = await client.options(
"/api/signal",
headers={
"Origin": _FRONTEND_ORIGIN,
"Access-Control-Request-Method": "POST",
"Access-Control-Request-Headers": "Content-Type, Authorization",
},
)
acao = resp.headers.get("access-control-allow-origin", "")
assert acao == _FRONTEND_ORIGIN, (
f"Ожидался Access-Control-Allow-Origin: {_FRONTEND_ORIGIN!r}, получен: {acao!r}"
)
@pytest.mark.asyncio
async def test_options_preflight_allow_methods_contains_get() -> None:
"""OPTIONS preflight должен вернуть Access-Control-Allow-Methods, включающий GET."""
async with make_app_client() as client:
resp = await client.options(
"/api/signal",
headers={
"Origin": _FRONTEND_ORIGIN,
"Access-Control-Request-Method": "GET",
"Access-Control-Request-Headers": "Authorization",
},
)
acam = resp.headers.get("access-control-allow-methods", "")
assert "GET" in acam, (
f"Access-Control-Allow-Methods не содержит GET: {acam!r}\n"
"Decision #1268: allow_methods=['POST'] — GET отсутствует"
)
@pytest.mark.asyncio
async def test_options_preflight_allow_headers_contains_authorization() -> None:
"""OPTIONS preflight должен вернуть Access-Control-Allow-Headers, включающий Authorization."""
async with make_app_client() as client:
resp = await client.options(
"/api/signal",
headers={
"Origin": _FRONTEND_ORIGIN,
"Access-Control-Request-Method": "POST",
"Access-Control-Request-Headers": "Authorization",
},
)
acah = resp.headers.get("access-control-allow-headers", "")
assert "authorization" in acah.lower(), (
f"Access-Control-Allow-Headers не содержит Authorization: {acah!r}"
)
@pytest.mark.asyncio
async def test_get_health_cors_header_present() -> None:
"""GET /health с Origin должен вернуть Access-Control-Allow-Origin (simple request)."""
async with make_app_client() as client:
resp = await client.get(
"/health",
headers={"Origin": _FRONTEND_ORIGIN},
)
assert resp.status_code == 200
acao = resp.headers.get("access-control-allow-origin", "")
assert acao == _FRONTEND_ORIGIN, (
f"GET /health: ожидался CORS-заголовок {_FRONTEND_ORIGIN!r}, получен: {acao!r}"
)

View file

@ -102,10 +102,10 @@ def test_register_request_rejects_old_placeholder_uuid(bad_uuid: str) -> None:
"agg-uuid-001",
"create-uuid-001",
])
def test_signal_request_rejects_old_placeholder_uuid(bad_uuid: str) -> None:
"""SignalRequest.user_id must reject old-style placeholder strings."""
with pytest.raises(ValidationError):
SignalRequest(user_id=bad_uuid, timestamp=1700000000000)
def test_signal_request_accepts_any_user_id_string(bad_uuid: str) -> None:
"""SignalRequest.user_id is optional (no pattern) — validation is at endpoint level."""
req = SignalRequest(user_id=bad_uuid, timestamp=1700000000000)
assert req.user_id == bad_uuid
# ---------------------------------------------------------------------------
@ -152,17 +152,16 @@ def test_register_request_rejects_uuid_v3_version_digit() -> None:
RegisterRequest(uuid="550e8400-e29b-31d4-a716-446655440000", name="Test")
def test_signal_request_rejects_uuid_wrong_variant_bits() -> None:
"""UUID with invalid variant bits (0xxx in fourth group) must be rejected."""
with pytest.raises(ValidationError):
# fourth group starts with '0' — not 8/9/a/b variant
SignalRequest(user_id="550e8400-e29b-41d4-0716-446655440000", timestamp=1700000000000)
def test_signal_request_accepts_any_variant_bits() -> None:
"""SignalRequest.user_id is now optional and unvalidated (JWT auth doesn't use it)."""
req = SignalRequest(user_id="550e8400-e29b-41d4-0716-446655440000", timestamp=1700000000000)
assert req.user_id is not None
def test_signal_request_rejects_uuid_wrong_variant_c() -> None:
"""UUID with variant 'c' (1100 bits) must be rejected — only 8/9/a/b allowed."""
with pytest.raises(ValidationError):
SignalRequest(user_id="550e8400-e29b-41d4-c716-446655440000", timestamp=1700000000000)
def test_signal_request_without_user_id() -> None:
"""SignalRequest works without user_id (JWT auth mode)."""
req = SignalRequest(timestamp=1700000000000)
assert req.user_id is None
def test_register_request_accepts_all_valid_v4_variants() -> None:

194
tests/test_fix_013.py Normal file
View file

@ -0,0 +1,194 @@
"""
Tests for BATON-FIX-013: CORS allow_methods добавить GET для /health эндпоинтов.
Acceptance criteria:
1. CORSMiddleware в main.py содержит "GET" в allow_methods.
2. OPTIONS preflight /health с Origin и Access-Control-Request-Method: GET
возвращает 200/204 и содержит GET в Access-Control-Allow-Methods.
3. OPTIONS preflight /api/health аналогично.
4. GET /health возвращает 200 (regression guard vs. allow_methods=['POST'] only).
"""
from __future__ import annotations
import os
os.environ.setdefault("BOT_TOKEN", "test-bot-token")
os.environ.setdefault("CHAT_ID", "-1001234567890")
os.environ.setdefault("WEBHOOK_SECRET", "test-webhook-secret")
os.environ.setdefault("WEBHOOK_URL", "https://example.com/api/webhook/telegram")
os.environ.setdefault("FRONTEND_ORIGIN", "http://localhost:3000")
os.environ.setdefault("ADMIN_TOKEN", "test-admin-token")
os.environ.setdefault("ADMIN_CHAT_ID", "5694335584")
import pytest
from tests.conftest import make_app_client
_ORIGIN = "http://localhost:3000"
# allow_headers = ["Content-Type", "Authorization"] — X-Custom-Header не разрешён,
# поэтому preflight с X-Custom-Header вернёт 400. Используем Content-Type.
_PREFLIGHT_HEADER = "Content-Type"
# ---------------------------------------------------------------------------
# Criterion 1 — Static: CORSMiddleware.allow_methods must contain "GET"
# ---------------------------------------------------------------------------
def test_cors_middleware_allow_methods_contains_get() -> None:
"""app.user_middleware CORSMiddleware должен содержать 'GET' в allow_methods."""
from fastapi.middleware.cors import CORSMiddleware
from backend.main import app
cors_mw = next(
(m for m in app.user_middleware if m.cls is CORSMiddleware), None
)
assert cors_mw is not None, "CORSMiddleware не найден в app.user_middleware"
allow_methods = cors_mw.kwargs.get("allow_methods", [])
assert "GET" in allow_methods, (
f"CORSMiddleware.allow_methods={allow_methods!r} не содержит 'GET'"
)
def test_cors_middleware_allow_methods_contains_head() -> None:
"""allow_methods должен содержать 'HEAD' для корректной работы preflight."""
from fastapi.middleware.cors import CORSMiddleware
from backend.main import app
cors_mw = next(
(m for m in app.user_middleware if m.cls is CORSMiddleware), None
)
assert cors_mw is not None
allow_methods = cors_mw.kwargs.get("allow_methods", [])
assert "HEAD" in allow_methods, (
f"CORSMiddleware.allow_methods={allow_methods!r} не содержит 'HEAD'"
)
def test_cors_middleware_allow_methods_contains_options() -> None:
"""allow_methods должен содержать 'OPTIONS' для корректной обработки preflight."""
from fastapi.middleware.cors import CORSMiddleware
from backend.main import app
cors_mw = next(
(m for m in app.user_middleware if m.cls is CORSMiddleware), None
)
assert cors_mw is not None
allow_methods = cors_mw.kwargs.get("allow_methods", [])
assert "OPTIONS" in allow_methods, (
f"CORSMiddleware.allow_methods={allow_methods!r} не содержит 'OPTIONS'"
)
# ---------------------------------------------------------------------------
# Criterion 2 — Preflight OPTIONS /health includes GET in Allow-Methods
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_health_preflight_options_returns_success_status() -> None:
"""OPTIONS preflight /health должен вернуть 200 или 204."""
async with make_app_client() as client:
response = await client.options(
"/health",
headers={
"Origin": _ORIGIN,
"Access-Control-Request-Method": "GET",
"Access-Control-Request-Headers": _PREFLIGHT_HEADER,
},
)
assert response.status_code in (200, 204), (
f"OPTIONS /health вернул {response.status_code}, ожидался 200 или 204"
)
@pytest.mark.asyncio
async def test_health_preflight_options_allow_methods_contains_get() -> None:
"""OPTIONS preflight /health: Access-Control-Allow-Methods должен содержать GET."""
async with make_app_client() as client:
response = await client.options(
"/health",
headers={
"Origin": _ORIGIN,
"Access-Control-Request-Method": "GET",
"Access-Control-Request-Headers": _PREFLIGHT_HEADER,
},
)
allow_methods_header = response.headers.get("access-control-allow-methods", "")
assert "GET" in allow_methods_header, (
f"Access-Control-Allow-Methods={allow_methods_header!r} не содержит 'GET'"
)
# ---------------------------------------------------------------------------
# Criterion 3 — Preflight OPTIONS /api/health includes GET in Allow-Methods
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_api_health_preflight_options_returns_success_status() -> None:
"""OPTIONS preflight /api/health должен вернуть 200 или 204."""
async with make_app_client() as client:
response = await client.options(
"/api/health",
headers={
"Origin": _ORIGIN,
"Access-Control-Request-Method": "GET",
"Access-Control-Request-Headers": _PREFLIGHT_HEADER,
},
)
assert response.status_code in (200, 204), (
f"OPTIONS /api/health вернул {response.status_code}, ожидался 200 или 204"
)
@pytest.mark.asyncio
async def test_api_health_preflight_options_allow_methods_contains_get() -> None:
"""OPTIONS preflight /api/health: Access-Control-Allow-Methods должен содержать GET."""
async with make_app_client() as client:
response = await client.options(
"/api/health",
headers={
"Origin": _ORIGIN,
"Access-Control-Request-Method": "GET",
"Access-Control-Request-Headers": _PREFLIGHT_HEADER,
},
)
allow_methods_header = response.headers.get("access-control-allow-methods", "")
assert "GET" in allow_methods_header, (
f"Access-Control-Allow-Methods={allow_methods_header!r} не содержит 'GET'"
)
# ---------------------------------------------------------------------------
# Criterion 4 — GET /health returns 200 (regression guard)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_health_get_returns_200_regression_guard() -> None:
"""GET /health должен вернуть 200 — regression guard против allow_methods=['POST'] only."""
async with make_app_client() as client:
response = await client.get(
"/health",
headers={"Origin": _ORIGIN},
)
assert response.status_code == 200, (
f"GET /health вернул {response.status_code}, ожидался 200"
)
@pytest.mark.asyncio
async def test_api_health_get_returns_200_regression_guard() -> None:
"""GET /api/health должен вернуть 200 — regression guard против allow_methods=['POST'] only."""
async with make_app_client() as client:
response = await client.get(
"/api/health",
headers={"Origin": _ORIGIN},
)
assert response.status_code == 200, (
f"GET /api/health вернул {response.status_code}, ожидался 200"
)

163
tests/test_fix_016.py Normal file
View file

@ -0,0 +1,163 @@
"""
Tests for BATON-FIX-016: VAPID public key убедиться, что ключ не вшит
как пустая строка в frontend-коде и читается через API.
Acceptance criteria:
1. В frontend-коде нет хардкода пустой строки в качестве VAPID key в <meta>-теге.
2. frontend читает ключ через API /api/vapid-public-key (_fetchVapidPublicKey).
3. GET /api/vapid-public-key возвращает HTTP 200.
4. GET /api/vapid-public-key возвращает JSON с полем vapid_public_key.
5. При наличии конфигурации VAPID_PUBLIC_KEY ответ содержит непустое значение.
"""
from __future__ import annotations
import os
import re
from pathlib import Path
from unittest.mock import patch
os.environ.setdefault("BOT_TOKEN", "test-bot-token")
os.environ.setdefault("CHAT_ID", "-1001234567890")
os.environ.setdefault("WEBHOOK_SECRET", "test-webhook-secret")
os.environ.setdefault("WEBHOOK_URL", "https://example.com/api/webhook/telegram")
os.environ.setdefault("FRONTEND_ORIGIN", "http://localhost:3000")
os.environ.setdefault("ADMIN_TOKEN", "test-admin-token")
os.environ.setdefault("ADMIN_CHAT_ID", "5694335584")
import pytest
from tests.conftest import make_app_client
PROJECT_ROOT = Path(__file__).parent.parent
FRONTEND_DIR = PROJECT_ROOT / "frontend"
INDEX_HTML = FRONTEND_DIR / "index.html"
APP_JS = FRONTEND_DIR / "app.js"
_TEST_VAPID_PUBLIC_KEY = "BFakeVapidPublicKeyForTestingPurposesOnlyBase64UrlEncoded"
# ---------------------------------------------------------------------------
# Criterion 1 — AST: no hardcoded empty VAPID key in <meta> tag (index.html)
# ---------------------------------------------------------------------------
def test_index_html_has_no_vapid_meta_tag_with_empty_content() -> None:
"""index.html не должен содержать <meta>-тег с application-server-key и пустым content."""
content = INDEX_HTML.read_text(encoding="utf-8")
match = re.search(
r'<meta[^>]*(?:application-server-key|vapid)[^>]*content\s*=\s*["\']["\']',
content,
re.IGNORECASE,
)
assert match is None, (
f"index.html содержит <meta>-тег с пустым VAPID ключом: {match.group(0)!r}"
)
def test_index_html_has_no_hardcoded_application_server_key_attribute() -> None:
"""index.html не должен содержать атрибут application-server-key вообще."""
content = INDEX_HTML.read_text(encoding="utf-8")
assert "application-server-key" not in content.lower(), (
"index.html содержит атрибут 'application-server-key'"
"VAPID ключ не должен быть вшит в HTML"
)
# ---------------------------------------------------------------------------
# Criterion 2 — AST: frontend reads key through API (app.js)
# ---------------------------------------------------------------------------
def test_app_js_contains_fetch_vapid_public_key_function() -> None:
"""app.js должен содержать функцию _fetchVapidPublicKey."""
content = APP_JS.read_text(encoding="utf-8")
assert "_fetchVapidPublicKey" in content, (
"app.js не содержит функцию _fetchVapidPublicKey — "
"чтение VAPID ключа через API не реализовано"
)
def test_app_js_fetch_vapid_calls_api_endpoint() -> None:
"""_fetchVapidPublicKey в app.js должна обращаться к /api/push/public-key (canonical URL)."""
content = APP_JS.read_text(encoding="utf-8")
assert "/api/push/public-key" in content, (
"app.js не содержит URL '/api/push/public-key' — VAPID ключ не читается через API"
)
def test_app_js_init_push_subscription_has_null_guard() -> None:
"""_initPushSubscription в app.js должна содержать guard против null ключа."""
content = APP_JS.read_text(encoding="utf-8")
assert re.search(r"if\s*\(\s*!\s*vapidPublicKey\s*\)", content), (
"app.js: _initPushSubscription не содержит guard 'if (!vapidPublicKey)'"
"подписка может быть создана без ключа"
)
def test_app_js_init_chains_fetch_vapid_then_init_subscription() -> None:
"""_init() в app.js должна вызывать _fetchVapidPublicKey().then(_initPushSubscription)."""
content = APP_JS.read_text(encoding="utf-8")
assert re.search(
r"_fetchVapidPublicKey\(\)\s*\.\s*then\s*\(\s*_initPushSubscription\s*\)",
content,
), (
"app.js: _init() не содержит цепочку _fetchVapidPublicKey().then(_initPushSubscription)"
)
def test_app_js_has_no_empty_string_hardcoded_as_application_server_key() -> None:
"""app.js не должен содержать хардкода пустой строки для applicationServerKey."""
content = APP_JS.read_text(encoding="utf-8")
match = re.search(r"applicationServerKey\s*[=:]\s*[\"']{2}", content)
assert match is None, (
f"app.js содержит хардкод пустой строки для applicationServerKey: {match.group(0)!r}"
)
# ---------------------------------------------------------------------------
# Criterion 3 — HTTP: GET /api/vapid-public-key returns 200
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_vapid_public_key_endpoint_returns_200() -> None:
"""GET /api/vapid-public-key должен вернуть HTTP 200."""
async with make_app_client() as client:
response = await client.get("/api/vapid-public-key")
assert response.status_code == 200, (
f"GET /api/vapid-public-key вернул {response.status_code}, ожидался 200"
)
# ---------------------------------------------------------------------------
# Criterion 4 — HTTP: response JSON contains vapid_public_key field
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_vapid_public_key_endpoint_returns_json_with_field() -> None:
"""GET /api/vapid-public-key должен вернуть JSON с полем vapid_public_key."""
async with make_app_client() as client:
response = await client.get("/api/vapid-public-key")
data = response.json()
assert "vapid_public_key" in data, (
f"Ответ /api/vapid-public-key не содержит поле 'vapid_public_key': {data!r}"
)
# ---------------------------------------------------------------------------
# Criterion 5 — HTTP: non-empty vapid_public_key when env var is configured
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_vapid_public_key_endpoint_returns_configured_value() -> None:
"""GET /api/vapid-public-key возвращает непустой ключ, когда VAPID_PUBLIC_KEY задан."""
with patch("backend.config.VAPID_PUBLIC_KEY", _TEST_VAPID_PUBLIC_KEY):
async with make_app_client() as client:
response = await client.get("/api/vapid-public-key")
data = response.json()
assert data.get("vapid_public_key") == _TEST_VAPID_PUBLIC_KEY, (
f"vapid_public_key должен быть '{_TEST_VAPID_PUBLIC_KEY}', "
f"получили: {data.get('vapid_public_key')!r}"
)

View file

@ -123,14 +123,16 @@ def test_signal_request_no_geo():
assert req.geo is None
def test_signal_request_missing_user_id():
with pytest.raises(ValidationError):
SignalRequest(timestamp=1742478000000) # type: ignore[call-arg]
def test_signal_request_without_user_id():
"""user_id is optional (JWT auth sends signals without it)."""
req = SignalRequest(timestamp=1742478000000)
assert req.user_id is None
def test_signal_request_empty_user_id():
with pytest.raises(ValidationError):
SignalRequest(user_id="", timestamp=1742478000000)
"""Empty string user_id is accepted (treated as None at endpoint level)."""
req = SignalRequest(user_id="", timestamp=1742478000000)
assert req.user_id == ""
def test_signal_request_timestamp_zero():

View file

@ -78,14 +78,14 @@ async def test_signal_without_geo_success():
@pytest.mark.asyncio
async def test_signal_missing_user_id_returns_422():
"""Missing user_id field must return 422."""
async def test_signal_missing_auth_returns_401():
"""Missing Authorization header must return 401."""
async with make_app_client() as client:
resp = await client.post(
"/api/signal",
json={"timestamp": 1742478000000},
)
assert resp.status_code == 422
assert resp.status_code == 401
@pytest.mark.asyncio

View file

@ -1,5 +1,5 @@
"""
Tests for backend/telegram.py: send_message, set_webhook, SignalAggregator.
Tests for backend/telegram.py: send_message, set_webhook, validate_bot_token.
NOTE: respx routes must be registered INSIDE the 'with mock:' block to be
intercepted properly. Registering them before entering the context does not
@ -25,8 +25,6 @@ def _safe_aiosqlite_await(self):
aiosqlite.core.Connection.__await__ = _safe_aiosqlite_await # type: ignore[method-assign]
import json
import os as _os
import tempfile
from unittest.mock import AsyncMock, patch
import httpx
@ -34,7 +32,7 @@ import pytest
import respx
from backend import config
from backend.telegram import SignalAggregator, send_message, set_webhook, validate_bot_token
from backend.telegram import send_message, set_webhook, validate_bot_token
SEND_URL = f"https://api.telegram.org/bot{config.BOT_TOKEN}/sendMessage"
@ -186,127 +184,6 @@ async def test_set_webhook_raises_on_non_200():
await set_webhook(url="https://example.com/webhook", secret="s")
# ---------------------------------------------------------------------------
# SignalAggregator helpers
# ---------------------------------------------------------------------------
async def _init_db_with_tmp() -> str:
"""Init a temp-file DB and return its path."""
from backend import config as _cfg, db as _db
path = tempfile.mktemp(suffix=".db")
_cfg.DB_PATH = path
await _db.init_db()
return path
def _cleanup(path: str) -> None:
for ext in ("", "-wal", "-shm"):
try:
_os.unlink(path + ext)
except FileNotFoundError:
pass
# ---------------------------------------------------------------------------
# SignalAggregator tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_aggregator_single_signal_calls_send_message():
"""Flushing an aggregator with one signal calls send_message once."""
path = await _init_db_with_tmp()
try:
agg = SignalAggregator(interval=9999)
await agg.add_signal(
user_uuid="a9900001-0000-4000-8000-000000000001",
user_name="Alice",
timestamp=1742478000000,
geo={"lat": 55.0, "lon": 37.0, "accuracy": 10.0},
signal_id=1,
)
with respx.mock(assert_all_called=False) as mock:
send_route = mock.post(SEND_URL).mock(
return_value=httpx.Response(200, json={"ok": True})
)
with patch("backend.telegram.db.save_telegram_batch", new_callable=AsyncMock):
with patch("backend.telegram.asyncio.sleep", new_callable=AsyncMock):
await agg.flush()
assert send_route.call_count == 1
finally:
_cleanup(path)
@pytest.mark.asyncio
async def test_aggregator_multiple_signals_one_message():
"""5 signals flushed at once produce exactly one send_message call."""
path = await _init_db_with_tmp()
try:
agg = SignalAggregator(interval=9999)
for i in range(5):
await agg.add_signal(
user_uuid=f"a990000{i}-0000-4000-8000-00000000000{i}",
user_name=f"User{i}",
timestamp=1742478000000 + i * 1000,
geo=None,
signal_id=i + 1,
)
with respx.mock(assert_all_called=False) as mock:
send_route = mock.post(SEND_URL).mock(
return_value=httpx.Response(200, json={"ok": True})
)
with patch("backend.telegram.db.save_telegram_batch", new_callable=AsyncMock):
with patch("backend.telegram.asyncio.sleep", new_callable=AsyncMock):
await agg.flush()
assert send_route.call_count == 1
finally:
_cleanup(path)
@pytest.mark.asyncio
async def test_aggregator_empty_buffer_no_send():
"""Flushing an empty aggregator must NOT call send_message."""
agg = SignalAggregator(interval=9999)
# No routes registered — if a POST is made it will raise AllMockedAssertionError
with respx.mock(assert_all_called=False) as mock:
send_route = mock.post(SEND_URL).mock(
return_value=httpx.Response(200, json={"ok": True})
)
await agg.flush()
assert send_route.call_count == 0
@pytest.mark.asyncio
async def test_aggregator_buffer_cleared_after_flush():
"""After flush, the aggregator buffer is empty."""
path = await _init_db_with_tmp()
try:
agg = SignalAggregator(interval=9999)
await agg.add_signal(
user_uuid="a9900099-0000-4000-8000-000000000099",
user_name="Test",
timestamp=1742478000000,
geo=None,
signal_id=99,
)
assert len(agg._buffer) == 1
with respx.mock(assert_all_called=False) as mock:
mock.post(SEND_URL).mock(return_value=httpx.Response(200, json={"ok": True}))
with patch("backend.telegram.db.save_telegram_batch", new_callable=AsyncMock):
with patch("backend.telegram.asyncio.sleep", new_callable=AsyncMock):
await agg.flush()
assert len(agg._buffer) == 0
finally:
_cleanup(path)
# ---------------------------------------------------------------------------
# BATON-007: 400 "chat not found" handling
# ---------------------------------------------------------------------------
@ -371,33 +248,3 @@ async def test_send_message_all_5xx_retries_exhausted_does_not_raise():
# Must not raise — message is dropped, service stays alive
await send_message("test all retries exhausted")
@pytest.mark.asyncio
async def test_aggregator_unknown_user_shows_uuid_prefix():
"""If user_name is None, the message shows first 8 chars of uuid."""
path = await _init_db_with_tmp()
try:
agg = SignalAggregator(interval=9999)
test_uuid = "abcdef1234567890"
await agg.add_signal(
user_uuid=test_uuid,
user_name=None,
timestamp=1742478000000,
geo=None,
signal_id=1,
)
sent_texts: list[str] = []
async def _fake_send(text: str) -> None:
sent_texts.append(text)
with patch("backend.telegram.send_message", side_effect=_fake_send):
with patch("backend.telegram.db.save_telegram_batch", new_callable=AsyncMock):
with patch("backend.telegram.asyncio.sleep", new_callable=AsyncMock):
await agg.flush()
assert len(sent_texts) == 1
assert test_uuid[:8] in sent_texts[0]
finally:
_cleanup(path)