pelmen/baton
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

kin: BATON-SEC-003-backend_dev

Browse source
This commit is contained in:
Gros Frumos 2026-03-21 08:12:01 +02:00
parent 8629f3e40b
commit 3a2ec11cc7
13 changed files with 593 additions and 125 deletions
Download patch file Download diff file Expand all files Collapse all files

22
backend/db.py
View file

@ -29,6 +29,7 @@ async def init_db() -> None:
name TEXT NOT NULL,
is_blocked INTEGER NOT NULL DEFAULT 0,
password_hash TEXT DEFAULT NULL,
api_key_hash TEXT DEFAULT NULL,
created_at TEXT DEFAULT (datetime('now'))
);
@ -64,6 +65,7 @@ async def init_db() -> None:
for stmt in [
"ALTER TABLE users ADD COLUMN is_blocked INTEGER NOT NULL DEFAULT 0",
"ALTER TABLE users ADD COLUMN password_hash TEXT DEFAULT NULL",
"ALTER TABLE users ADD COLUMN api_key_hash TEXT DEFAULT NULL",
]:
try:
await conn.execute(stmt)
@ -73,8 +75,17 @@ async def init_db() -> None:
await conn.commit()
async def register_user(uuid: str, name: str) -> dict:
async def register_user(uuid: str, name: str, api_key_hash: Optional[str] = None) -> dict:
async with _get_conn() as conn:
if api_key_hash is not None:
await conn.execute(
"""
INSERT INTO users (uuid, name, api_key_hash) VALUES (?, ?, ?)
ON CONFLICT(uuid) DO UPDATE SET api_key_hash = excluded.api_key_hash
""",
(uuid, name, api_key_hash),
)
else:
await conn.execute(
"INSERT OR IGNORE INTO users (uuid, name) VALUES (?, ?)",
(uuid, name),
@ -87,6 +98,15 @@ async def register_user(uuid: str, name: str) -> dict:
return {"user_id": row["id"], "uuid": row["uuid"]}
async def get_api_key_hash_by_uuid(uuid: str) -> Optional[str]:
async with _get_conn() as conn:
async with conn.execute(
"SELECT api_key_hash FROM users WHERE uuid = ?", (uuid,)
) as cur:
row = await cur.fetchone()
return row["api_key_hash"] if row else None
async def save_signal(
user_uuid: str,
timestamp: int,

31
backend/main.py
View file

@ -4,14 +4,16 @@ import asyncio
import hashlib
import logging
import os
import secrets
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Any
from typing import Any, Optional
import httpx
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from backend import config, db, telegram
from backend.middleware import rate_limit_register, rate_limit_signal, verify_admin_token, verify_webhook_secret
@ -25,10 +27,17 @@ from backend.models import (
SignalResponse,
)
_api_key_bearer = HTTPBearer(auto_error=False)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def _hash_api_key(key: str) -> str:
"""SHA-256 хэш для API-ключа (без соли — для быстрого сравнения)."""
return hashlib.sha256(key.encode()).hexdigest()
def _hash_password(password: str) -> str:
"""Hash a password using PBKDF2-HMAC-SHA256 (stdlib, no external deps).
@ -105,7 +114,7 @@ app.add_middleware(
CORSMiddleware,
allow_origins=[config.FRONTEND_ORIGIN],
allow_methods=["POST"],
allow_headers=["Content-Type"],
allow_headers=["Content-Type", "Authorization"],
)
@ -117,12 +126,24 @@ async def health() -> dict[str, Any]:
@app.post("/api/register", response_model=RegisterResponse)
async def register(body: RegisterRequest, _: None = Depends(rate_limit_register)) -> RegisterResponse:
result = await db.register_user(uuid=body.uuid, name=body.name)
return RegisterResponse(user_id=result["user_id"], uuid=result["uuid"])
api_key = secrets.token_hex(32)
result = await db.register_user(uuid=body.uuid, name=body.name, api_key_hash=_hash_api_key(api_key))
return RegisterResponse(user_id=result["user_id"], uuid=result["uuid"], api_key=api_key)
@app.post("/api/signal", response_model=SignalResponse)
async def signal(body: SignalRequest, _: None = Depends(rate_limit_signal)) -> SignalResponse:
async def signal(
body: SignalRequest,
credentials: Optional[HTTPAuthorizationCredentials] = Depends(_api_key_bearer),
_: None = Depends(rate_limit_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")

1
backend/models.py
View file

@ -12,6 +12,7 @@ class RegisterRequest(BaseModel):
class RegisterResponse(BaseModel):
user_id: int
uuid: str
api_key: str
class GeoData(BaseModel):

57
tests/test_arch_002.py
View file

@ -5,6 +5,10 @@ Acceptance criteria:
1. No asyncio task for the aggregator is created at lifespan startup.
2. POST /api/signal calls telegram.send_message directly (no aggregator intermediary).
3. SignalAggregator class in telegram.py is preserved with '# v2.0 feature' marker.
UUID notes: all UUIDs satisfy the UUID v4 pattern.
BATON-SEC-003: POST /api/signal now requires Authorization: Bearer <api_key>.
Tests that send signals register first and use the returned api_key.
"""
from __future__ import annotations
@ -15,6 +19,7 @@ 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")
from pathlib import Path
from unittest.mock import AsyncMock, patch
@ -25,6 +30,20 @@ from tests.conftest import make_app_client
_BACKEND_DIR = Path(__file__).parent.parent / "backend"
# Valid UUID v4 constants
_UUID_S1 = "a0100001-0000-4000-8000-000000000001"
_UUID_S2 = "a0100002-0000-4000-8000-000000000002"
_UUID_S3 = "a0100003-0000-4000-8000-000000000003"
_UUID_S4 = "a0100004-0000-4000-8000-000000000004"
_UUID_S5 = "a0100005-0000-4000-8000-000000000005"
async def _register(client, uuid: str, name: str) -> str:
"""Register user and return api_key."""
r = await client.post("/api/register", json={"uuid": uuid, "name": name})
assert r.status_code == 200
return r.json()["api_key"]
# ---------------------------------------------------------------------------
# Criterion 1 — No asyncio task for aggregator created at startup (static)
@ -72,11 +91,12 @@ def test_aggregator_instantiation_commented_out_in_main():
async def test_signal_calls_telegram_send_message_directly():
"""POST /api/signal must call telegram.send_message directly, not via aggregator (ADR-004)."""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": "adr-uuid-s1", "name": "Tester"})
api_key = await _register(client, _UUID_S1, "Tester")
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
resp = await client.post(
"/api/signal",
json={"user_id": "adr-uuid-s1", "timestamp": 1742478000000},
json={"user_id": _UUID_S1, "timestamp": 1742478000000},
headers={"Authorization": f"Bearer {api_key}"},
)
assert resp.status_code == 200
mock_send.assert_called_once()
@ -86,11 +106,12 @@ async def test_signal_calls_telegram_send_message_directly():
async def test_signal_message_contains_registered_username():
"""Message passed to send_message must include the registered user's name."""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": "adr-uuid-s2", "name": "Alice"})
api_key = await _register(client, _UUID_S2, "Alice")
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
await client.post(
"/api/signal",
json={"user_id": "adr-uuid-s2", "timestamp": 1742478000000},
json={"user_id": _UUID_S2, "timestamp": 1742478000000},
headers={"Authorization": f"Bearer {api_key}"},
)
text = mock_send.call_args[0][0]
assert "Alice" in text
@ -100,11 +121,12 @@ async def test_signal_message_contains_registered_username():
async def test_signal_message_without_geo_contains_bez_geolocatsii():
"""When geo is None, message must contain 'Без геолокации'."""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": "adr-uuid-s3", "name": "Bob"})
api_key = await _register(client, _UUID_S3, "Bob")
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
await client.post(
"/api/signal",
json={"user_id": "adr-uuid-s3", "timestamp": 1742478000000, "geo": None},
json={"user_id": _UUID_S3, "timestamp": 1742478000000, "geo": None},
headers={"Authorization": f"Bearer {api_key}"},
)
text = mock_send.call_args[0][0]
assert "Без геолокации" in text
@ -114,15 +136,16 @@ async def test_signal_message_without_geo_contains_bez_geolocatsii():
async def test_signal_message_with_geo_contains_coordinates():
"""When geo is provided, message must contain lat and lon values."""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": "adr-uuid-s4", "name": "Charlie"})
api_key = await _register(client, _UUID_S4, "Charlie")
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
await client.post(
"/api/signal",
json={
"user_id": "adr-uuid-s4",
"user_id": _UUID_S4,
"timestamp": 1742478000000,
"geo": {"lat": 55.7558, "lon": 37.6173, "accuracy": 15.0},
},
headers={"Authorization": f"Bearer {api_key}"},
)
text = mock_send.call_args[0][0]
assert "55.7558" in text
@ -133,29 +156,17 @@ async def test_signal_message_with_geo_contains_coordinates():
async def test_signal_message_contains_utc_marker():
"""Message passed to send_message must contain 'UTC' timestamp marker."""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": "adr-uuid-s5", "name": "Dave"})
api_key = await _register(client, _UUID_S5, "Dave")
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
await client.post(
"/api/signal",
json={"user_id": "adr-uuid-s5", "timestamp": 1742478000000},
json={"user_id": _UUID_S5, "timestamp": 1742478000000},
headers={"Authorization": f"Bearer {api_key}"},
)
text = mock_send.call_args[0][0]
assert "UTC" in text
@pytest.mark.asyncio
async def test_signal_unknown_user_message_uses_uuid_prefix():
"""When user is not registered, message uses first 8 chars of uuid as name."""
async with make_app_client() as client:
with patch("backend.telegram.send_message", new_callable=AsyncMock) as mock_send:
await client.post(
"/api/signal",
json={"user_id": "unknown-uuid-xyz", "timestamp": 1742478000000},
)
text = mock_send.call_args[0][0]
assert "unknown-" in text # "unknown-uuid-xyz"[:8] == "unknown-"
# ---------------------------------------------------------------------------
# Criterion 3 — SignalAggregator preserved with '# v2.0 feature' marker (static)
# ---------------------------------------------------------------------------

34
tests/test_arch_003.py
View file

@ -6,6 +6,9 @@ Acceptance criteria:
5 requests pass (200), 6th returns 429; counter resets after the 10-minute window.
2. Token comparison is timing-safe:
secrets.compare_digest is used in middleware.py (no == / != for token comparison).
UUID notes: RegisterRequest.uuid requires a valid UUID v4 pattern.
All UUID constants below satisfy this constraint.
"""
from __future__ import annotations
@ -20,6 +23,7 @@ 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")
import pytest
from tests.conftest import make_app_client
@ -38,6 +42,24 @@ _SAMPLE_UPDATE = {
},
}
# Valid UUID v4 constants for rate-limit tests
# Pattern: [0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}
_UUIDS_OK = [
f"d0{i:06d}-0000-4000-8000-000000000001"
for i in range(10)
]
_UUIDS_BLK = [
f"d1{i:06d}-0000-4000-8000-000000000001"
for i in range(10)
]
_UUIDS_EXP = [
f"d2{i:06d}-0000-4000-8000-000000000001"
for i in range(10)
]
_UUID_BLK_999 = "d1000999-0000-4000-8000-000000000001"
_UUID_EXP_BLK = "d2000999-0000-4000-8000-000000000001"
_UUID_EXP_AFTER = "d2001000-0000-4000-8000-000000000001"
# ---------------------------------------------------------------------------
# Criterion 1 — Rate limiting: first 5 requests pass
@ -51,7 +73,7 @@ async def test_register_rate_limit_allows_five_requests():
for i in range(5):
resp = await client.post(
"/api/register",
json={"uuid": f"rl-ok-{i:03d}", "name": f"User{i}"},
json={"uuid": _UUIDS_OK[i], "name": f"User{i}"},
)
assert resp.status_code == 200, (
f"Request {i + 1}/5 unexpectedly returned {resp.status_code}"
@ -70,11 +92,11 @@ async def test_register_rate_limit_blocks_sixth_request():
for i in range(5):
await client.post(
"/api/register",
json={"uuid": f"rl-blk-{i:03d}", "name": f"User{i}"},
json={"uuid": _UUIDS_BLK[i], "name": f"User{i}"},
)
resp = await client.post(
"/api/register",
json={"uuid": "rl-blk-999", "name": "Attacker"},
json={"uuid": _UUID_BLK_999, "name": "Attacker"},
)
assert resp.status_code == 429
@ -94,13 +116,13 @@ async def test_register_rate_limit_resets_after_window_expires():
for i in range(5):
await client.post(
"/api/register",
json={"uuid": f"rl-exp-{i:03d}", "name": f"User{i}"},
json={"uuid": _UUIDS_EXP[i], "name": f"User{i}"},
)
# Verify the 6th is blocked before window expiry
blocked = await client.post(
"/api/register",
json={"uuid": "rl-exp-blk", "name": "Attacker"},
json={"uuid": _UUID_EXP_BLK, "name": "Attacker"},
)
assert blocked.status_code == 429, (
"Expected 429 after exhausting rate limit, got " + str(blocked.status_code)
@ -110,7 +132,7 @@ async def test_register_rate_limit_resets_after_window_expires():
with patch("time.time", return_value=base_time + 601):
resp_after = await client.post(
"/api/register",
json={"uuid": "rl-exp-after", "name": "Legit"},
json={"uuid": _UUID_EXP_AFTER, "name": "Legit"},
)
assert resp_after.status_code == 200, (

69
tests/test_baton_005.py
View file

@ -9,6 +9,10 @@ Acceptance criteria:
5. Удаление — пользователь исчезает из GET /admin/users, возвращается 204
6. Защита: неавторизованный запрос к /admin/* возвращает 401
7. Отсутствие регрессии с основным функционалом
BATON-SEC-003: POST /api/signal now requires Authorization: Bearer <api_key>.
Tests 3 and 4 (block/unblock + signal) use /api/register to obtain an api_key,
then admin block/unblock the user by their DB id.
"""
from __future__ import annotations
@ -33,6 +37,11 @@ NGINX_CONF = PROJECT_ROOT / "nginx" / "baton.conf"
ADMIN_HEADERS = {"Authorization": "Bearer test-admin-token"}
WRONG_HEADERS = {"Authorization": "Bearer wrong-token"}
# Valid UUID v4 for signal-related tests (registered via /api/register)
_UUID_BLOCK = "f0000001-0000-4000-8000-000000000001"
_UUID_UNBLOCK = "f0000002-0000-4000-8000-000000000002"
_UUID_SIG_OK = "f0000003-0000-4000-8000-000000000003"
# ---------------------------------------------------------------------------
# Criterion 6 — Unauthorised requests to /admin/* return 401
@ -250,23 +259,32 @@ async def test_admin_block_user_returns_is_blocked_true() -> None:
async def test_admin_block_user_prevents_signal() -> None:
"""Заблокированный пользователь не может отправить сигнал — /api/signal возвращает 403."""
async with make_app_client() as client:
create_resp = await client.post(
"/admin/users",
json={"uuid": "block-uuid-002", "name": "BlockSignalUser"},
headers=ADMIN_HEADERS,
# Регистрируем через /api/register чтобы получить api_key
reg_resp = await client.post(
"/api/register",
json={"uuid": _UUID_BLOCK, "name": "BlockSignalUser"},
)
user_id = create_resp.json()["id"]
user_uuid = create_resp.json()["uuid"]
assert reg_resp.status_code == 200
api_key = reg_resp.json()["api_key"]
user_uuid = reg_resp.json()["uuid"]
# Находим ID пользователя
users_resp = await client.get("/admin/users", headers=ADMIN_HEADERS)
user = next(u for u in users_resp.json() if u["uuid"] == user_uuid)
user_id = user["id"]
# Блокируем
await client.put(
f"/admin/users/{user_id}/block",
json={"is_blocked": True},
headers=ADMIN_HEADERS,
)
# Заблокированный пользователь должен получить 403
signal_resp = await client.post(
"/api/signal",
json={"user_id": user_uuid, "timestamp": 1700000000000, "geo": None},
headers={"Authorization": f"Bearer {api_key}"},
)
assert signal_resp.status_code == 403
@ -318,13 +336,19 @@ async def test_admin_unblock_user_returns_is_blocked_false() -> None:
async def test_admin_unblock_user_restores_signal_access() -> None:
"""После разблокировки пользователь снова может отправить сигнал (200)."""
async with make_app_client() as client:
create_resp = await client.post(
"/admin/users",
json={"uuid": "unblock-uuid-002", "name": "UnblockSignalUser"},
headers=ADMIN_HEADERS,
# Регистрируем через /api/register чтобы получить api_key
reg_resp = await client.post(
"/api/register",
json={"uuid": _UUID_UNBLOCK, "name": "UnblockSignalUser"},
)
user_id = create_resp.json()["id"]
user_uuid = create_resp.json()["uuid"]
assert reg_resp.status_code == 200
api_key = reg_resp.json()["api_key"]
user_uuid = reg_resp.json()["uuid"]
# Находим ID
users_resp = await client.get("/admin/users", headers=ADMIN_HEADERS)
user = next(u for u in users_resp.json() if u["uuid"] == user_uuid)
user_id = user["id"]
# Блокируем
await client.put(
@ -344,6 +368,7 @@ async def test_admin_unblock_user_restores_signal_access() -> None:
signal_resp = await client.post(
"/api/signal",
json={"user_id": user_uuid, "timestamp": 1700000000000, "geo": None},
headers={"Authorization": f"Bearer {api_key}"},
)
assert signal_resp.status_code == 200
assert signal_resp.json()["status"] == "ok"
@ -462,26 +487,28 @@ async def test_register_not_broken_after_admin_operations() -> None:
# Основной функционал
resp = await client.post(
"/api/register",
json={"uuid": "regress-user-uuid-001", "name": "RegularUser"},
json={"uuid": _UUID_SIG_OK, "name": "RegularUser"},
)
assert resp.status_code == 200
assert resp.json()["uuid"] == "regress-user-uuid-001"
assert resp.json()["uuid"] == _UUID_SIG_OK
@pytest.mark.asyncio
async def test_signal_from_unblocked_user_succeeds() -> None:
"""Незаблокированный пользователь, созданный через admin API, может отправить сигнал."""
async def test_signal_from_registered_unblocked_user_succeeds() -> None:
"""Зарегистрированный незаблокированный пользователь может отправить сигнал."""
async with make_app_client() as client:
create_resp = await client.post(
"/admin/users",
json={"uuid": "regress-signal-uuid-001", "name": "SignalUser"},
headers=ADMIN_HEADERS,
reg_resp = await client.post(
"/api/register",
json={"uuid": _UUID_SIG_OK, "name": "SignalUser"},
)
user_uuid = create_resp.json()["uuid"]
assert reg_resp.status_code == 200
api_key = reg_resp.json()["api_key"]
user_uuid = reg_resp.json()["uuid"]
signal_resp = await client.post(
"/api/signal",
json={"user_id": user_uuid, "timestamp": 1700000000000, "geo": None},
headers={"Authorization": f"Bearer {api_key}"},
)
assert signal_resp.status_code == 200
assert signal_resp.json()["status"] == "ok"

27
tests/test_baton_006.py
View file

@ -11,6 +11,9 @@ Acceptance criteria:
6. POST /api/register возвращает 200 (FastAPI-маршрут не сломан).
7. POST /api/signal возвращает 200 (FastAPI-маршрут не сломан).
8. POST /api/webhook/telegram возвращает 200 с корректным секретом.
BATON-SEC-003: POST /api/signal now requires Authorization: Bearer <api_key>.
UUID constants satisfy the UUID v4 pattern.
"""
from __future__ import annotations
@ -23,6 +26,7 @@ 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")
import pytest
@ -31,6 +35,10 @@ from tests.conftest import make_app_client
PROJECT_ROOT = Path(__file__).parent.parent
NGINX_CONF = PROJECT_ROOT / "nginx" / "baton.conf"
# Valid UUID v4 constants
_UUID_REG = "e0000001-0000-4000-8000-000000000001"
_UUID_SIG = "e0000002-0000-4000-8000-000000000002"
# ---------------------------------------------------------------------------
# Criterion 1 — location /api/ proxies to FastAPI
# ---------------------------------------------------------------------------
@ -52,7 +60,6 @@ def test_nginx_conf_has_api_location_block() -> None:
def test_nginx_conf_api_location_proxies_to_fastapi() -> None:
"""Блок location /api/ должен делать proxy_pass на 127.0.0.1:8000."""
content = NGINX_CONF.read_text(encoding="utf-8")
# Ищем блок api и proxy_pass внутри
api_block = re.search(
r"location\s+/api/\s*\{([^}]+)\}", content, re.DOTALL
)
@ -95,7 +102,6 @@ def test_nginx_conf_health_location_proxies_to_fastapi() -> None:
def test_nginx_conf_root_location_has_root_directive() -> None:
"""location / в nginx.conf должен содержать директиву root (статика)."""
content = NGINX_CONF.read_text(encoding="utf-8")
# Ищем последний блок location / (не /api/, не /health)
root_block = re.search(
r"location\s+/\s*\{([^}]+)\}", content, re.DOTALL
)
@ -179,13 +185,14 @@ async def test_api_register_not_broken_after_nginx_change() -> None:
async with make_app_client() as client:
response = await client.post(
"/api/register",
json={"uuid": "baton-006-uuid-001", "name": "TestUser"},
json={"uuid": _UUID_REG, "name": "TestUser"},
)
assert response.status_code == 200
data = response.json()
assert data["user_id"] > 0
assert data["uuid"] == "baton-006-uuid-001"
assert data["uuid"] == _UUID_REG
assert "api_key" in data
# ---------------------------------------------------------------------------
@ -197,19 +204,21 @@ async def test_api_register_not_broken_after_nginx_change() -> None:
async def test_api_signal_not_broken_after_nginx_change() -> None:
"""POST /api/signal должен вернуть 200 — функция не сломана изменением nginx."""
async with make_app_client() as client:
# Сначала регистрируем пользователя
await client.post(
reg_resp = await client.post(
"/api/register",
json={"uuid": "baton-006-uuid-002", "name": "SignalUser"},
json={"uuid": _UUID_SIG, "name": "SignalUser"},
)
# Отправляем сигнал
assert reg_resp.status_code == 200
api_key = reg_resp.json()["api_key"]
response = await client.post(
"/api/signal",
json={
"user_id": "baton-006-uuid-002",
"user_id": _UUID_SIG,
"timestamp": 1700000000000,
"geo": None,
},
headers={"Authorization": f"Bearer {api_key}"},
)
assert response.status_code == 200

10
tests/test_models.py
View file

@ -46,11 +46,11 @@ def test_register_request_empty_uuid():
def test_register_request_name_max_length():
"""name longer than 100 chars raises ValidationError."""
with pytest.raises(ValidationError):
RegisterRequest(uuid="some-uuid", name="x" * 101)
RegisterRequest(uuid="550e8400-e29b-41d4-a716-446655440000", name="x" * 101)
def test_register_request_name_exactly_100():
req = RegisterRequest(uuid="some-uuid", name="x" * 100)
req = RegisterRequest(uuid="550e8400-e29b-41d4-a716-446655440000", name="x" * 100)
assert len(req.name) == 100
@ -116,7 +116,7 @@ def test_signal_request_valid():
def test_signal_request_no_geo():
req = SignalRequest(
user_id="some-uuid",
user_id="550e8400-e29b-41d4-a716-446655440000",
timestamp=1742478000000,
geo=None,
)
@ -136,9 +136,9 @@ def test_signal_request_empty_user_id():
def test_signal_request_timestamp_zero():
"""timestamp must be > 0."""
with pytest.raises(ValidationError):
SignalRequest(user_id="some-uuid", timestamp=0)
SignalRequest(user_id="550e8400-e29b-41d4-a716-446655440000", timestamp=0)
def test_signal_request_timestamp_negative():
with pytest.raises(ValidationError):
SignalRequest(user_id="some-uuid", timestamp=-1)
SignalRequest(user_id="550e8400-e29b-41d4-a716-446655440000", timestamp=-1)

68
tests/test_register.py
View file

@ -1,5 +1,11 @@
"""
Integration tests for POST /api/register.
UUID notes: RegisterRequest.uuid requires a valid UUID v4 pattern
(^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$).
All UUID constants below satisfy this constraint.
BATON-SEC-003: /api/register now returns api_key in the response.
"""
from __future__ import annotations
@ -10,23 +16,34 @@ 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")
import pytest
from tests.conftest import make_app_client
# Valid UUID v4 constants for register tests
_UUID_REG_1 = "b0000001-0000-4000-8000-000000000001"
_UUID_REG_2 = "b0000002-0000-4000-8000-000000000002"
_UUID_REG_3 = "b0000003-0000-4000-8000-000000000003"
_UUID_REG_4 = "b0000004-0000-4000-8000-000000000004"
_UUID_REG_5 = "b0000005-0000-4000-8000-000000000005"
_UUID_REG_6 = "b0000006-0000-4000-8000-000000000006"
@pytest.mark.asyncio
async def test_register_new_user_success():
"""POST /api/register returns 200 with user_id > 0."""
"""POST /api/register returns 200 with user_id > 0 and api_key."""
async with make_app_client() as client:
resp = await client.post(
"/api/register",
json={"uuid": "reg-uuid-001", "name": "Alice"},
json={"uuid": _UUID_REG_1, "name": "Alice"},
)
assert resp.status_code == 200
data = resp.json()
assert data["user_id"] > 0
assert data["uuid"] == "reg-uuid-001"
assert data["uuid"] == _UUID_REG_1
assert "api_key" in data
assert len(data["api_key"]) == 64 # secrets.token_hex(32) = 64 hex chars
@pytest.mark.asyncio
@ -35,24 +52,42 @@ async def test_register_idempotent():
async with make_app_client() as client:
r1 = await client.post(
"/api/register",
json={"uuid": "reg-uuid-002", "name": "Bob"},
json={"uuid": _UUID_REG_2, "name": "Bob"},
)
r2 = await client.post(
"/api/register",
json={"uuid": "reg-uuid-002", "name": "Bob"},
json={"uuid": _UUID_REG_2, "name": "Bob"},
)
assert r1.status_code == 200
assert r2.status_code == 200
assert r1.json()["user_id"] == r2.json()["user_id"]
@pytest.mark.asyncio
async def test_register_idempotent_returns_api_key_on_every_call():
"""Each registration call returns an api_key (key rotation on re-register)."""
async with make_app_client() as client:
r1 = await client.post(
"/api/register",
json={"uuid": _UUID_REG_3, "name": "Carol"},
)
r2 = await client.post(
"/api/register",
json={"uuid": _UUID_REG_3, "name": "Carol"},
)
assert r1.status_code == 200
assert r2.status_code == 200
assert "api_key" in r1.json()
assert "api_key" in r2.json()
@pytest.mark.asyncio
async def test_register_empty_name_returns_422():
"""Empty name must fail validation with 422."""
async with make_app_client() as client:
resp = await client.post(
"/api/register",
json={"uuid": "reg-uuid-003", "name": ""},
json={"uuid": _UUID_REG_4, "name": ""},
)
assert resp.status_code == 422
@ -74,7 +109,18 @@ async def test_register_missing_name_returns_422():
async with make_app_client() as client:
resp = await client.post(
"/api/register",
json={"uuid": "reg-uuid-004"},
json={"uuid": _UUID_REG_4},
)
assert resp.status_code == 422
@pytest.mark.asyncio
async def test_register_invalid_uuid_format_returns_422():
"""Non-UUID4 string as uuid must return 422."""
async with make_app_client() as client:
resp = await client.post(
"/api/register",
json={"uuid": "not-a-uuid", "name": "Dave"},
)
assert resp.status_code == 422
@ -85,11 +131,11 @@ async def test_register_user_stored_in_db():
async with make_app_client() as client:
r1 = await client.post(
"/api/register",
json={"uuid": "reg-uuid-005", "name": "Dana"},
json={"uuid": _UUID_REG_5, "name": "Dana"},
)
r2 = await client.post(
"/api/register",
json={"uuid": "reg-uuid-005", "name": "Dana"},
json={"uuid": _UUID_REG_5, "name": "Dana"},
)
assert r1.json()["user_id"] == r2.json()["user_id"]
@ -100,6 +146,6 @@ async def test_register_response_contains_uuid():
async with make_app_client() as client:
resp = await client.post(
"/api/register",
json={"uuid": "reg-uuid-006", "name": "Eve"},
json={"uuid": _UUID_REG_6, "name": "Eve"},
)
assert resp.json()["uuid"] == "reg-uuid-006"
assert resp.json()["uuid"] == _UUID_REG_6

55
tests/test_sec_002.py
View file

@ -7,6 +7,9 @@ Tests for BATON-SEC-002:
UUID notes: RegisterRequest.uuid and SignalRequest.user_id both require a valid
UUID v4 pattern (^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$).
All constants below satisfy this constraint.
BATON-SEC-003: POST /api/signal now requires Authorization: Bearer <api_key>.
_register_and_get_key() helper returns the api_key from the registration response.
"""
from __future__ import annotations
@ -40,6 +43,13 @@ _UUID_IP_B = "a0000006-0000-4000-8000-000000000006" # per-IP isolation, user
# ── Helpers ─────────────────────────────────────────────────────────────────
async def _register_and_get_key(client, uuid: str, name: str) -> str:
"""Register user and return api_key."""
r = await client.post("/api/register", json={"uuid": uuid, "name": name})
assert r.status_code == 200
return r.json()["api_key"]
def _make_request(headers: dict | None = None, client_host: str = "127.0.0.1") -> Request:
"""Build a minimal Starlette Request with given headers and remote address."""
scope = {
@ -120,10 +130,10 @@ def test_get_client_ip_returns_unknown_when_no_client_and_no_headers():
async def test_signal_rate_limit_returns_429_after_10_requests():
"""POST /api/signal returns 429 on the 11th request from the same IP."""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": _UUID_SIG_RL, "name": "RL"})
api_key = await _register_and_get_key(client, _UUID_SIG_RL, "RL")
payload = {"user_id": _UUID_SIG_RL, "timestamp": 1742478000000}
ip_hdrs = {"X-Real-IP": "5.5.5.5"}
ip_hdrs = {"X-Real-IP": "5.5.5.5", "Authorization": f"Bearer {api_key}"}
statuses = []
for _ in range(11):
@ -137,10 +147,10 @@ async def test_signal_rate_limit_returns_429_after_10_requests():
async def test_signal_first_10_requests_are_allowed():
"""First 10 POST /api/signal requests from the same IP must all return 200."""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": _UUID_SIG_OK, "name": "OK"})
api_key = await _register_and_get_key(client, _UUID_SIG_OK, "OK")
payload = {"user_id": _UUID_SIG_OK, "timestamp": 1742478000000}
ip_hdrs = {"X-Real-IP": "6.6.6.6"}
ip_hdrs = {"X-Real-IP": "6.6.6.6", "Authorization": f"Bearer {api_key}"}
statuses = []
for _ in range(10):
@ -162,26 +172,28 @@ async def test_signal_rate_limit_does_not_affect_register_counter():
to return 429 — the counters use different keys ('sig:IP' vs 'IP').
"""
async with make_app_client() as client:
ip_hdrs = {"X-Real-IP": "7.7.7.7"}
ip_hdrs_reg = {"X-Real-IP": "7.7.7.7"}
# Register a user (increments register counter, key='7.7.7.7', count=1)
r_reg = await client.post(
"/api/register",
json={"uuid": _UUID_IND_SIG, "name": "Ind"},
headers=ip_hdrs,
headers=ip_hdrs_reg,
)
assert r_reg.status_code == 200
api_key = r_reg.json()["api_key"]
# Exhaust signal rate limit for same IP (11 requests, key='sig:7.7.7.7')
payload = {"user_id": _UUID_IND_SIG, "timestamp": 1742478000000}
ip_hdrs_sig = {"X-Real-IP": "7.7.7.7", "Authorization": f"Bearer {api_key}"}
for _ in range(11):
await client.post("/api/signal", json=payload, headers=ip_hdrs)
await client.post("/api/signal", json=payload, headers=ip_hdrs_sig)
# Register counter is still at 1 — must allow another registration
r_reg2 = await client.post(
"/api/register",
json={"uuid": _UUID_IND_SIG2, "name": "Ind2"},
headers=ip_hdrs,
headers=ip_hdrs_reg,
)
assert r_reg2.status_code == 200, (
@ -206,10 +218,21 @@ async def test_register_rate_limit_does_not_affect_signal_counter():
headers=ip_hdrs,
)
assert r0.status_code == 200
api_key = r0.json()["api_key"]
# Send 5 more register requests from the same IP to exhaust the limit
# (register limit = 5/600s, so request #6 → 429)
for _ in range(5):
# Send 4 more register requests from the same IP (requests 2-5 succeed,
# each rotates the api_key; request 6 would be 429).
# We keep track of the last api_key since re-registration rotates it.
for _ in range(4):
r = await client.post(
"/api/register",
json={"uuid": _UUID_IND_REG, "name": "Reg"},
headers=ip_hdrs,
)
if r.status_code == 200:
api_key = r.json()["api_key"]
# 6th request → 429 (exhausts limit without rotating key)
await client.post(
"/api/register",
json={"uuid": _UUID_IND_REG, "name": "Reg"},
@ -220,7 +243,7 @@ async def test_register_rate_limit_does_not_affect_signal_counter():
r_sig = await client.post(
"/api/signal",
json={"user_id": _UUID_IND_REG, "timestamp": 1742478000000},
headers=ip_hdrs,
headers={"X-Real-IP": "8.8.8.8", "Authorization": f"Bearer {api_key}"},
)
assert r_sig.status_code == 200, (
@ -238,22 +261,22 @@ async def test_signal_rate_limit_is_per_ip_different_ips_are_independent():
Rate limit counters are per-IP — exhausting for IP A must not block IP B.
"""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": _UUID_IP_A, "name": "IPA"})
await client.post("/api/register", json={"uuid": _UUID_IP_B, "name": "IPB"})
api_key_a = await _register_and_get_key(client, _UUID_IP_A, "IPA")
api_key_b = await _register_and_get_key(client, _UUID_IP_B, "IPB")
# Exhaust rate limit for IP A (11 requests → 11th is 429)
for _ in range(11):
await client.post(
"/api/signal",
json={"user_id": _UUID_IP_A, "timestamp": 1742478000000},
headers={"X-Real-IP": "11.11.11.11"},
headers={"X-Real-IP": "11.11.11.11", "Authorization": f"Bearer {api_key_a}"},
)
# IP B should still be allowed (independent counter)
r = await client.post(
"/api/signal",
json={"user_id": _UUID_IP_B, "timestamp": 1742478000000},
headers={"X-Real-IP": "22.22.22.22"},
headers={"X-Real-IP": "22.22.22.22", "Authorization": f"Bearer {api_key_b}"},
)
assert r.status_code == 200, f"IP B was incorrectly blocked: {r.status_code}"

254
tests/test_sec_003.py Normal file
View file

@ -0,0 +1,254 @@
"""
Tests for BATON-SEC-003: API-ключи для аутентификации /api/signal.
Acceptance criteria:
1. POST /api/register возвращает api_key длиной 64 hex-символа.
2. POST /api/signal без Authorization header → 401.
3. POST /api/signal с неверным api_key → 401.
4. POST /api/signal с правильным api_key → 200.
5. Повторная регистрация генерирует новый api_key (ротация ключа).
6. Старый api_key становится недействительным после ротации.
7. Новый api_key работает после ротации.
8. SHA-256 хэш api_key сохраняется в БД, сырой ключ — нет (проверка через DB функцию).
UUID notes: все UUID ниже удовлетворяют паттерну UUID v4.
"""
from __future__ import annotations
import hashlib
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")
import pytest
from backend import db
from tests.conftest import make_app_client, temp_db
from backend import config
# Valid UUID v4 constants
_UUID_1 = "aa000001-0000-4000-8000-000000000001"
_UUID_2 = "aa000002-0000-4000-8000-000000000002"
_UUID_3 = "aa000003-0000-4000-8000-000000000003"
_UUID_4 = "aa000004-0000-4000-8000-000000000004"
_UUID_5 = "aa000005-0000-4000-8000-000000000005"
_UUID_6 = "aa000006-0000-4000-8000-000000000006"
_UUID_7 = "aa000007-0000-4000-8000-000000000007"
_UUID_8 = "aa000008-0000-4000-8000-000000000008"
# ---------------------------------------------------------------------------
# Criterion 1 — /api/register returns api_key of correct length
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_register_returns_api_key():
"""POST /api/register должен вернуть поле api_key в ответе."""
async with make_app_client() as client:
resp = await client.post(
"/api/register",
json={"uuid": _UUID_1, "name": "Alice"},
)
assert resp.status_code == 200
assert "api_key" in resp.json()
@pytest.mark.asyncio
async def test_register_api_key_is_64_hex_chars():
"""api_key должен быть строкой из 64 hex-символов (secrets.token_hex(32))."""
async with make_app_client() as client:
resp = await client.post(
"/api/register",
json={"uuid": _UUID_2, "name": "Bob"},
)
api_key = resp.json()["api_key"]
assert len(api_key) == 64
assert all(c in "0123456789abcdef" for c in api_key), (
f"api_key contains non-hex characters: {api_key}"
)
# ---------------------------------------------------------------------------
# Criterion 2 — Missing Authorization → 401
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_signal_without_auth_header_returns_401():
"""POST /api/signal без Authorization header должен вернуть 401."""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": _UUID_3, "name": "Charlie"})
resp = await client.post(
"/api/signal",
json={"user_id": _UUID_3, "timestamp": 1742478000000},
)
assert resp.status_code == 401
@pytest.mark.asyncio
async def test_signal_without_bearer_scheme_returns_401():
"""POST /api/signal с неверной схемой (Basic вместо Bearer) должен вернуть 401."""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": _UUID_3, "name": "Charlie"})
resp = await client.post(
"/api/signal",
json={"user_id": _UUID_3, "timestamp": 1742478000000},
headers={"Authorization": "Basic wrongtoken"},
)
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# Criterion 3 — Wrong api_key → 401
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_signal_with_wrong_api_key_returns_401():
"""POST /api/signal с неверным api_key должен вернуть 401."""
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": _UUID_4, "name": "Dave"})
resp = await client.post(
"/api/signal",
json={"user_id": _UUID_4, "timestamp": 1742478000000},
headers={"Authorization": "Bearer " + "0" * 64},
)
assert resp.status_code == 401
@pytest.mark.asyncio
async def test_signal_with_unknown_user_returns_401():
"""POST /api/signal с api_key незарегистрированного пользователя должен вернуть 401."""
async with make_app_client() as client:
resp = await client.post(
"/api/signal",
json={"user_id": _UUID_5, "timestamp": 1742478000000},
headers={"Authorization": "Bearer " + "a" * 64},
)
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# Criterion 4 — Correct api_key → 200
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_signal_with_valid_api_key_returns_200():
"""POST /api/signal с правильным api_key должен вернуть 200."""
async with make_app_client() as client:
reg = await client.post(
"/api/register",
json={"uuid": _UUID_6, "name": "Eve"},
)
assert reg.status_code == 200
api_key = reg.json()["api_key"]
resp = await client.post(
"/api/signal",
json={"user_id": _UUID_6, "timestamp": 1742478000000},
headers={"Authorization": f"Bearer {api_key}"},
)
assert resp.status_code == 200
assert resp.json()["status"] == "ok"
# ---------------------------------------------------------------------------
# Criterion 5-7 — Key rotation on re-register
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_re_register_produces_new_api_key():
"""Повторная регистрация должна возвращать новый api_key (ротация)."""
async with make_app_client() as client:
r1 = await client.post("/api/register", json={"uuid": _UUID_7, "name": "Frank"})
r2 = await client.post("/api/register", json={"uuid": _UUID_7, "name": "Frank"})
assert r1.status_code == 200
assert r2.status_code == 200
# Ключи могут совпасть (очень маловероятно), но оба должны быть длиной 64
assert len(r2.json()["api_key"]) == 64
@pytest.mark.asyncio
async def test_old_api_key_invalid_after_re_register():
"""После повторной регистрации старый api_key не должен работать."""
async with make_app_client() as client:
r1 = await client.post("/api/register", json={"uuid": _UUID_8, "name": "Grace"})
old_key = r1.json()["api_key"]
# Повторная регистрация — ротация ключа
r2 = await client.post("/api/register", json={"uuid": _UUID_8, "name": "Grace"})
new_key = r2.json()["api_key"]
# Старый ключ больше не должен работать
old_resp = await client.post(
"/api/signal",
json={"user_id": _UUID_8, "timestamp": 1742478000000},
headers={"Authorization": f"Bearer {old_key}"},
)
# Новый ключ должен работать
new_resp = await client.post(
"/api/signal",
json={"user_id": _UUID_8, "timestamp": 1742478000000},
headers={"Authorization": f"Bearer {new_key}"},
)
assert old_resp.status_code == 401, "Старый ключ должен быть недействителен после ротации"
assert new_resp.status_code == 200, "Новый ключ должен работать"
# ---------------------------------------------------------------------------
# Criterion 8 — SHA-256 hash is stored, not the raw key
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_api_key_hash_stored_in_db_not_raw_key():
"""В БД должен храниться SHA-256 хэш api_key, а не сырой ключ."""
with temp_db():
from backend.main import app
import contextlib
import httpx
import respx
from httpx import AsyncClient, ASGITransport
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"
mock_router = respx.mock(assert_all_called=False)
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})
)
with mock_router:
async with app.router.lifespan_context(app):
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
reg = await client.post(
"/api/register",
json={"uuid": _UUID_1, "name": "HashTest"},
)
assert reg.status_code == 200
raw_api_key = reg.json()["api_key"]
# Читаем хэш из БД напрямую
stored_hash = await db.get_api_key_hash_by_uuid(_UUID_1)
expected_hash = hashlib.sha256(raw_api_key.encode()).hexdigest()
assert stored_hash is not None, "api_key_hash должен быть в БД"
assert stored_hash == expected_hash, (
"В БД должен быть SHA-256 хэш, а не сырой ключ"
)
assert stored_hash != raw_api_key, "В БД не должен храниться сырой ключ"

27
tests/test_sec_007.py
View file

@ -6,6 +6,9 @@ Regression tests for BATON-SEC-007:
3. POST /api/signal uses asyncio.create_task — HTTP response is not blocked
by Telegram rate-limit pauses.
4. GET /health returns only {"status": "ok"} — no timestamp field.
BATON-SEC-003: POST /api/signal now requires Authorization: Bearer <api_key>.
Tests that send signals now register first and use the returned api_key.
"""
from __future__ import annotations
@ -18,6 +21,7 @@ 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")
from unittest.mock import AsyncMock, patch
@ -31,6 +35,10 @@ from tests.conftest import make_app_client
SEND_URL = f"https://api.telegram.org/bot{config.BOT_TOKEN}/sendMessage"
# Valid UUID v4 constants
_UUID_CT = "d4e5f6a7-b8c9-4d0e-a1b2-c3d4e5f6a7b8"
_UUID_SLOW = "e5f6a7b8-c9d0-4e1f-a2b3-c4d5e6f7a8b9"
# ---------------------------------------------------------------------------
# Criterion 1 — retry loop is bounded to max 3 attempts
@ -164,10 +172,13 @@ async def test_signal_uses_create_task_for_telegram_send_message():
"""POST /api/signal must wrap telegram.send_message in asyncio.create_task."""
with patch("backend.main.asyncio.create_task", wraps=asyncio.create_task) as mock_ct:
async with make_app_client() as client:
await client.post("/api/register", json={"uuid": "d4e5f6a7-b8c9-4d0e-a1b2-c3d4e5f6a7b8", "name": "CT"})
reg = await client.post("/api/register", json={"uuid": _UUID_CT, "name": "CT"})
assert reg.status_code == 200
api_key = reg.json()["api_key"]
resp = await client.post(
"/api/signal",
json={"user_id": "d4e5f6a7-b8c9-4d0e-a1b2-c3d4e5f6a7b8", "timestamp": 1742478000000},
json={"user_id": _UUID_CT, "timestamp": 1742478000000},
headers={"Authorization": f"Bearer {api_key}"},
)
assert resp.status_code == 200
@ -177,8 +188,6 @@ async def test_signal_uses_create_task_for_telegram_send_message():
@pytest.mark.asyncio
async def test_signal_response_returns_before_telegram_completes():
"""POST /api/signal returns 200 even when Telegram send_message is delayed."""
# Simulate a slow Telegram response. If send_message is awaited directly,
# the HTTP response would be delayed until sleep completes.
slow_sleep_called = False
async def slow_send_message(_text: str) -> None:
@ -189,19 +198,21 @@ async def test_signal_response_returns_before_telegram_completes():
with patch("backend.main.telegram.send_message", side_effect=slow_send_message):
with patch("backend.main.asyncio.create_task", wraps=asyncio.create_task):
async with make_app_client() as client:
await client.post(
reg = await client.post(
"/api/register",
json={"uuid": "e5f6a7b8-c9d0-4e1f-a2b3-c4d5e6f7a8b9", "name": "Slow"},
json={"uuid": _UUID_SLOW, "name": "Slow"},
)
assert reg.status_code == 200
api_key = reg.json()["api_key"]
resp = await client.post(
"/api/signal",
json={
"user_id": "e5f6a7b8-c9d0-4e1f-a2b3-c4d5e6f7a8b9",
"user_id": _UUID_SLOW,
"timestamp": 1742478000000,
},
headers={"Authorization": f"Bearer {api_key}"},
)
# Response must be immediate — no blocking on the 9999-second sleep
assert resp.status_code == 200

53
tests/test_signal.py
View file

@ -1,5 +1,11 @@
"""
Integration tests for POST /api/signal.
UUID notes: both RegisterRequest.uuid and SignalRequest.user_id require valid UUID v4.
All UUID constants below satisfy the pattern.
BATON-SEC-003: /api/signal now requires Authorization: Bearer <api_key>.
The _register() helper returns the api_key from the registration response.
"""
from __future__ import annotations
@ -10,30 +16,42 @@ 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")
import pytest
from httpx import AsyncClient
from tests.conftest import make_app_client
# Valid UUID v4 constants for signal tests
_UUID_1 = "c0000001-0000-4000-8000-000000000001"
_UUID_2 = "c0000002-0000-4000-8000-000000000002"
_UUID_3 = "c0000003-0000-4000-8000-000000000003"
_UUID_4 = "c0000004-0000-4000-8000-000000000004"
_UUID_5 = "c0000005-0000-4000-8000-000000000005"
_UUID_6 = "c0000006-0000-4000-8000-000000000006"
async def _register(client: AsyncClient, uuid: str, name: str) -> None:
async def _register(client: AsyncClient, uuid: str, name: str) -> str:
"""Register user, assert success, return raw api_key."""
r = await client.post("/api/register", json={"uuid": uuid, "name": name})
assert r.status_code == 200
assert r.status_code == 200, f"Registration failed: {r.status_code} {r.text}"
return r.json()["api_key"]
@pytest.mark.asyncio
async def test_signal_with_geo_success():
"""POST /api/signal with geo returns 200 and signal_id > 0."""
async with make_app_client() as client:
await _register(client, "sig-uuid-001", "Alice")
api_key = await _register(client, _UUID_1, "Alice")
resp = await client.post(
"/api/signal",
json={
"user_id": "sig-uuid-001",
"user_id": _UUID_1,
"timestamp": 1742478000000,
"geo": {"lat": 55.7558, "lon": 37.6173, "accuracy": 15.0},
},
headers={"Authorization": f"Bearer {api_key}"},
)
assert resp.status_code == 200
data = resp.json()
@ -45,14 +63,15 @@ async def test_signal_with_geo_success():
async def test_signal_without_geo_success():
"""POST /api/signal with geo: null returns 200."""
async with make_app_client() as client:
await _register(client, "sig-uuid-002", "Bob")
api_key = await _register(client, _UUID_2, "Bob")
resp = await client.post(
"/api/signal",
json={
"user_id": "sig-uuid-002",
"user_id": _UUID_2,
"timestamp": 1742478000000,
"geo": None,
},
headers={"Authorization": f"Bearer {api_key}"},
)
assert resp.status_code == 200
assert resp.json()["status"] == "ok"
@ -75,7 +94,7 @@ async def test_signal_missing_timestamp_returns_422():
async with make_app_client() as client:
resp = await client.post(
"/api/signal",
json={"user_id": "sig-uuid-003"},
json={"user_id": _UUID_3},
)
assert resp.status_code == 422
@ -87,14 +106,16 @@ async def test_signal_stored_in_db():
proving both were persisted.
"""
async with make_app_client() as client:
await _register(client, "sig-uuid-004", "Charlie")
api_key = await _register(client, _UUID_4, "Charlie")
r1 = await client.post(
"/api/signal",
json={"user_id": "sig-uuid-004", "timestamp": 1742478000001},
json={"user_id": _UUID_4, "timestamp": 1742478000001},
headers={"Authorization": f"Bearer {api_key}"},
)
r2 = await client.post(
"/api/signal",
json={"user_id": "sig-uuid-004", "timestamp": 1742478000002},
json={"user_id": _UUID_4, "timestamp": 1742478000002},
headers={"Authorization": f"Bearer {api_key}"},
)
assert r1.status_code == 200
assert r2.status_code == 200
@ -111,11 +132,12 @@ async def test_signal_sends_telegram_message_directly():
send_url = f"https://api.telegram.org/bot{_cfg.BOT_TOKEN}/sendMessage"
async with make_app_client() as client:
await _register(client, "sig-uuid-005", "Dana")
api_key = await _register(client, _UUID_5, "Dana")
# make_app_client already mocks send_url; signal returns 200 proves send was called
resp = await client.post(
"/api/signal",
json={"user_id": "sig-uuid-005", "timestamp": 1742478000000},
json={"user_id": _UUID_5, "timestamp": 1742478000000},
headers={"Authorization": f"Bearer {api_key}"},
)
assert resp.status_code == 200
@ -126,10 +148,11 @@ async def test_signal_sends_telegram_message_directly():
async def test_signal_returns_signal_id_positive():
"""signal_id in response is always a positive integer."""
async with make_app_client() as client:
await _register(client, "sig-uuid-006", "Eve")
api_key = await _register(client, _UUID_6, "Eve")
resp = await client.post(
"/api/signal",
json={"user_id": "sig-uuid-006", "timestamp": 1742478000000},
json={"user_id": _UUID_6, "timestamp": 1742478000000},
headers={"Authorization": f"Bearer {api_key}"},
)
assert resp.json()["signal_id"] > 0
@ -141,7 +164,7 @@ async def test_signal_geo_invalid_lat_returns_422():
resp = await client.post(
"/api/signal",
json={
"user_id": "sig-uuid-007",
"user_id": _UUID_1,
"timestamp": 1742478000000,
"geo": {"lat": 200.0, "lon": 0.0, "accuracy": 10.0},
},