596 lines
34 KiB
Python
596 lines
34 KiB
Python
"""Тесты безопасности: аутентификация, права доступа, изоляция данных, токены, загрузка файлов, инъекции.
|
|
|
|
Тесты с пометкой xfail фиксируют ИЗВЕСТНЫЕ недостатки: когда недостаток устранят,
|
|
тест начнёт проходить (XPASS) — тогда пометку нужно снять.
|
|
"""
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
os.environ.setdefault("DATABASE_URL", os.environ.get("TEST_DATABASE_URL") or "sqlite:///" + os.path.join(tempfile.mkdtemp(), "s.db").replace("\\", "/"))
|
|
os.environ.setdefault("SCHEDULER_INTERVAL", "3600")
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-pytest-0123456789abcdef")
|
|
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
|
|
|
import jwt # noqa: E402
|
|
from fastapi.routing import APIRoute # noqa: E402
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
from app import config # noqa: E402
|
|
from app.main import app # noqa: E402
|
|
from app.routers import catalog # noqa: E402
|
|
|
|
V1 = "/api/v1"
|
|
PNG = b"\x89PNG\r\n\x1a\n" + b"\0" * 64
|
|
|
|
|
|
def bearer(t):
|
|
return {"Authorization": "Bearer " + t}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_limiter():
|
|
catalog._fails.clear()
|
|
yield
|
|
catalog._fails.clear()
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def env():
|
|
"""Один набор пользователей/данных на модуль: админ, модератор, инженер, два техника (на смене), выходной техник."""
|
|
with TestClient(app) as c:
|
|
def login(u, p):
|
|
r = c.post(f"{V1}/auth/login", json={"username": u, "password": p})
|
|
assert r.status_code == 200, r.text
|
|
return r.json()
|
|
|
|
adm = bearer(login("admin", "admin")["access_token"])
|
|
always = c.post(f"{V1}/schedules", headers=adm, json={"name": "sec-always", "kind": "weekly",
|
|
"weekdays": [0, 1, 2, 3, 4, 5, 6], "start_time": "00:00:00", "end_time": "23:59:00"}).json()
|
|
tech = c.post(f"{V1}/positions", headers=adm, json={"name": "sec-tech", "kind": "technician",
|
|
"schedule_id": always["id"]}).json()
|
|
eng = c.post(f"{V1}/positions", headers=adm, json={"name": "sec-eng", "kind": "engineer"}).json()
|
|
ids = {}
|
|
for name, pos, extra in (("sec_a", tech, {}), ("sec_b", tech, {}), ("sec_eng", eng, {}),
|
|
("sec_mod", None, {"is_moderator": True})):
|
|
r = c.post(f"{V1}/users", headers=adm, json={"username": name, "password": "Passw0rd!x",
|
|
"position_id": pos["id"] if pos else None, **extra})
|
|
assert r.status_code == 201, r.text
|
|
ids[name] = r.json()["id"]
|
|
h = {n: bearer(login(n, "Passw0rd!x")["access_token"]) for n in ids}
|
|
h["admin"] = adm
|
|
eng_h = h["sec_eng"]
|
|
loc = c.post(f"{V1}/locations", headers=eng_h, json={"name": "sec-loc"}).json()
|
|
eq = c.post(f"{V1}/equipment", headers=eng_h, json={"name": "sec-eq", "location_id": loc["id"]}).json()
|
|
cl = c.post(f"{V1}/checklists", headers=eng_h, json={"name": "sec-cl", "items": [
|
|
{"text": "Осмотр", "answer_type": "status", "photo": "optional"},
|
|
{"text": "Давление", "answer_type": "number", "min_value": 1, "max_value": 5}]}).json()
|
|
order = c.post(f"{V1}/orders", headers=eng_h, json={
|
|
"title": "sec-order", "checklist_id": cl["id"], "equipment_id": eq["id"],
|
|
"assignee_user_id": ids["sec_a"], "reviewer_user_id": ids["sec_eng"]}).json()
|
|
yield {"c": c, "h": h, "ids": ids, "order": order, "cl": cl, "eq": eq, "login": login}
|
|
|
|
|
|
# ---------------------------------------------------------------- аутентификация
|
|
PUBLIC = {("POST", "/auth/login"), ("POST", "/auth/refresh"), ("POST", "/auth/logout"),
|
|
("GET", "/health"), ("GET", "/demo/status"),
|
|
("GET", "/public/report-options"), ("POST", "/public/incidents"), ("GET", "/app/version"),
|
|
("GET", "/docs/raw/{token}")} # исходник для ONLYOFFICE: только по короткой подписанной ссылке (проверка — test_scope_docs) # публичная форма инцидента: без входа, с лимитом
|
|
|
|
|
|
def test_every_api_route_requires_auth(env):
|
|
"""Любой маршрут API без токена даёт 401 (кроме явно публичных)."""
|
|
c, checked = env["c"], 0
|
|
for r in app.routes:
|
|
if not isinstance(r, APIRoute) or not r.path.startswith(V1 + "/"):
|
|
continue
|
|
sub = r.path[len(V1):]
|
|
for m in r.methods - {"HEAD", "OPTIONS"}:
|
|
if (m, sub) in PUBLIC:
|
|
continue
|
|
path = r.path.replace("{id}", "1").replace("{item_id}", "1")
|
|
resp = c.request(m, path, json={} if m in ("POST", "PUT", "PATCH") else None)
|
|
assert resp.status_code == 401, f"{m} {path} -> {resp.status_code} без токена"
|
|
checked += 1
|
|
assert checked > 50
|
|
|
|
|
|
def test_public_endpoints_leak_nothing(env):
|
|
c = env["c"]
|
|
assert c.get(f"{V1}/health").json().keys() == {"ok", "api_version", "time", "demo"}
|
|
assert c.get(f"{V1}/demo/status").json() == {"active": False}
|
|
|
|
|
|
def test_forged_and_invalid_tokens(env):
|
|
c, uid = env["c"], env["ids"]["sec_a"]
|
|
exp = datetime.now(timezone.utc) + timedelta(minutes=5)
|
|
good = {"sub": str(uid), "exp": exp, "typ": "access", "mode": "real"}
|
|
cases = {
|
|
"чужой ключ": jwt.encode(good, "wrong-secret", algorithm="HS256"),
|
|
"alg=none": jwt.encode(good, None, algorithm="none"),
|
|
"просрочен": jwt.encode({**good, "exp": datetime.now(timezone.utc) - timedelta(minutes=1)}, config.SECRET_KEY, algorithm="HS256"),
|
|
"режим demo в боевом": jwt.encode({**good, "mode": "demo"}, config.SECRET_KEY, algorithm="HS256"),
|
|
"sub не число": jwt.encode({**good, "sub": "1 OR 1=1"}, config.SECRET_KEY, algorithm="HS256"),
|
|
"нет пользователя": jwt.encode({**good, "sub": "999999"}, config.SECRET_KEY, algorithm="HS256"),
|
|
"мусор": "abc.def.ghi",
|
|
}
|
|
for name, tok in cases.items():
|
|
assert c.get(f"{V1}/me", headers=bearer(tok)).status_code == 401, name
|
|
# контроль: правильно подписанный токен работает
|
|
assert c.get(f"{V1}/me", headers=bearer(jwt.encode(good, config.SECRET_KEY, algorithm="HS256"))).status_code == 200
|
|
# схема не Bearer
|
|
assert c.get(f"{V1}/me", headers={"Authorization": "Basic YWRtaW46YWRtaW4="}).status_code == 401
|
|
|
|
|
|
def test_refresh_token_is_not_an_access_token(env):
|
|
c = env["c"]
|
|
t = env["login"]("sec_a", "Passw0rd!x")
|
|
assert c.get(f"{V1}/me", headers=bearer(t["refresh_token"])).status_code == 401
|
|
|
|
|
|
def test_login_errors_do_not_enumerate_users(env):
|
|
c = env["c"]
|
|
a = c.post(f"{V1}/auth/login", json={"username": "sec_a", "password": "bad"})
|
|
b = c.post(f"{V1}/auth/login", json={"username": "no_such_user", "password": "bad"})
|
|
assert a.status_code == b.status_code == 401 and a.json() == b.json()
|
|
|
|
|
|
def test_login_rate_limit(env):
|
|
c = env["c"]
|
|
codes = [c.post(f"{V1}/auth/login", json={"username": "sec_b", "password": "x"}).status_code for _ in range(7)]
|
|
assert codes[:config.LOGIN_MAX_FAILS] == [401] * config.LOGIN_MAX_FAILS
|
|
assert codes[-1] == 429
|
|
# блокировка действует и на верный пароль, пока не истечёт окно
|
|
r = c.post(f"{V1}/auth/login", json={"username": "sec_b", "password": "Passw0rd!x"})
|
|
assert r.status_code == 429 and "Retry-After" in r.headers
|
|
|
|
|
|
def test_deactivated_user_is_locked_out_immediately(env):
|
|
c, h = env["c"], env["h"]
|
|
r = c.post(f"{V1}/users", headers=h["admin"], json={"username": "sec_tmp", "password": "Passw0rd!x"})
|
|
uid = r.json()["id"]
|
|
t = env["login"]("sec_tmp", "Passw0rd!x")
|
|
assert c.get(f"{V1}/me", headers=bearer(t["access_token"])).status_code == 200
|
|
assert c.patch(f"{V1}/users/{uid}", headers=h["admin"], json={"is_active": False}).status_code == 200
|
|
assert c.get(f"{V1}/me", headers=bearer(t["access_token"])).status_code == 401 # access
|
|
assert c.post(f"{V1}/auth/refresh", json={"refresh_token": t["refresh_token"]}).status_code == 401 # refresh
|
|
assert c.post(f"{V1}/auth/login", json={"username": "sec_tmp", "password": "Passw0rd!x"}).status_code == 401
|
|
|
|
|
|
def test_refresh_rotation_and_reuse_detection(env):
|
|
c = env["c"]
|
|
t1 = env["login"]("sec_b", "Passw0rd!x")
|
|
t2 = c.post(f"{V1}/auth/refresh", json={"refresh_token": t1["refresh_token"]}).json()
|
|
assert t2["refresh_token"] != t1["refresh_token"]
|
|
# повторное предъявление старого — отзыв ВСЕХ сессий пользователя
|
|
assert c.post(f"{V1}/auth/refresh", json={"refresh_token": t1["refresh_token"]}).status_code == 401
|
|
assert c.post(f"{V1}/auth/refresh", json={"refresh_token": t2["refresh_token"]}).status_code == 401
|
|
|
|
|
|
def test_logout_revokes_refresh(env):
|
|
c = env["c"]
|
|
t = env["login"]("sec_b", "Passw0rd!x")
|
|
assert c.post(f"{V1}/auth/logout", json={"refresh_token": t["refresh_token"]}).status_code == 204
|
|
assert c.post(f"{V1}/auth/refresh", json={"refresh_token": t["refresh_token"]}).status_code == 401
|
|
|
|
|
|
def test_password_change_and_reset_kill_other_sessions(env):
|
|
c, h = env["c"], env["h"]
|
|
uid = c.post(f"{V1}/users", headers=h["admin"], json={"username": "sec_pw", "password": "Passw0rd!x"}).json()["id"]
|
|
t = env["login"]("sec_pw", "Passw0rd!x")
|
|
other = env["login"]("sec_pw", "Passw0rd!x")
|
|
n = c.post(f"{V1}/auth/change-password", headers=bearer(t["access_token"]),
|
|
json={"old_password": "Passw0rd!x", "new_password": "N3wPassw0rd!"})
|
|
assert n.status_code == 200
|
|
assert c.post(f"{V1}/auth/refresh", json={"refresh_token": other["refresh_token"]}).status_code == 401
|
|
assert c.post(f"{V1}/auth/login", json={"username": "sec_pw", "password": "Passw0rd!x"}).status_code == 401
|
|
# сброс пароля администратором тоже завершает сессии
|
|
cur = c.post(f"{V1}/auth/login", json={"username": "sec_pw", "password": "N3wPassw0rd!"}).json()
|
|
assert c.patch(f"{V1}/users/{uid}", headers=h["admin"], json={"password": "Res3tPassw0rd!"}).status_code == 200
|
|
assert c.post(f"{V1}/auth/refresh", json={"refresh_token": cur["refresh_token"]}).status_code == 401
|
|
|
|
|
|
def test_change_password_requires_old_password(env):
|
|
c, h = env["c"], env["h"]
|
|
r = c.post(f"{V1}/auth/change-password", headers=h["sec_b"], json={"old_password": "wrong", "new_password": "N3wPassw0rd!"})
|
|
assert r.status_code == 403
|
|
|
|
|
|
def test_password_hash_never_exposed(env):
|
|
c, h = env["c"], env["h"]
|
|
for path in ("/me", "/users"):
|
|
body = c.get(V1 + path, headers=h["sec_a"]).text
|
|
assert "password" not in body.lower() and "$2b$" not in body
|
|
|
|
|
|
# ---------------------------------------------------------------- права (RBAC)
|
|
def test_technician_cannot_use_management_endpoints(env):
|
|
c, t = env["c"], env["h"]["sec_a"]
|
|
denied = [
|
|
("POST", "/users", {"username": "hack", "password": "Passw0rd!x"}),
|
|
("PATCH", f"/users/{env['ids']['sec_b']}", {"full_name": "x"}),
|
|
("DELETE", f"/users/{env['ids']['sec_b']}", None),
|
|
("POST", "/positions", {"name": "x", "kind": "technician"}),
|
|
("POST", "/schedules", {"name": "x", "kind": "weekly", "weekdays": [0]}),
|
|
("POST", "/locations", {"name": "x"}),
|
|
("POST", "/equipment", {"name": "x"}),
|
|
("POST", "/checklists", {"name": "x", "items": []}),
|
|
("GET", "/order-templates", None),
|
|
("POST", "/orders", {"title": "x", "checklist_id": env["cl"]["id"]}),
|
|
("POST", f"/orders/{env['order']['id']}/cancel", None),
|
|
("GET", "/settings", None),
|
|
("PUT", "/settings", {"max_photo_mb": 99}),
|
|
("GET", "/reports", None),
|
|
("GET", "/reports/completed-orders?from=2026-01-01T00:00&to=2026-02-01T00:00", None),
|
|
("POST", "/demo/enable", {"confirm": True}),
|
|
("GET", "/demo", None),
|
|
]
|
|
for m, p, body in denied:
|
|
r = c.request(m, V1 + p, headers=t, json=body)
|
|
assert r.status_code == 403, f"{m} {p} -> {r.status_code}"
|
|
|
|
|
|
def test_engineer_cannot_manage_users_or_settings(env):
|
|
c, e = env["c"], env["h"]["sec_eng"]
|
|
assert c.post(f"{V1}/users", headers=e, json={"username": "hack2", "password": "Passw0rd!x"}).status_code == 403
|
|
assert c.post(f"{V1}/positions", headers=e, json={"name": "x", "kind": "technician"}).status_code == 403
|
|
assert c.get(f"{V1}/settings", headers=e).status_code == 403
|
|
|
|
|
|
def test_moderator_cannot_escalate(env):
|
|
c, m, ids = env["c"], env["h"]["sec_mod"], env["ids"]
|
|
admin_id = c.get(f"{V1}/me", headers=env["h"]["admin"]).json()["id"]
|
|
assert c.get(f"{V1}/settings", headers=m).status_code == 403
|
|
# выдать права
|
|
assert c.post(f"{V1}/users", headers=m, json={"username": "x_adm", "password": "Passw0rd!x", "is_admin": True}).status_code == 403
|
|
assert c.post(f"{V1}/users", headers=m, json={"username": "x_mod", "password": "Passw0rd!x", "is_moderator": True}).status_code == 403
|
|
assert c.patch(f"{V1}/users/{ids['sec_a']}", headers=m, json={"is_admin": True}).status_code == 403
|
|
assert c.patch(f"{V1}/users/{ids['sec_mod']}", headers=m, json={"is_admin": True}).status_code == 403 # себе
|
|
# трогать администратора: сменить пароль, отключить, удалить
|
|
assert c.patch(f"{V1}/users/{admin_id}", headers=m, json={"password": "Hacked1234!"}).status_code == 403
|
|
assert c.patch(f"{V1}/users/{admin_id}", headers=m, json={"is_active": False}).status_code == 403
|
|
assert c.delete(f"{V1}/users/{admin_id}", headers=m).status_code == 403
|
|
# пароль администратора не изменился
|
|
assert c.post(f"{V1}/auth/login", json={"username": "admin", "password": "admin"}).status_code == 200
|
|
|
|
|
|
def test_cannot_delete_self(env):
|
|
c, a = env["c"], env["h"]["admin"]
|
|
me = c.get(f"{V1}/me", headers=a).json()["id"]
|
|
assert c.delete(f"{V1}/users/{me}", headers=a).status_code == 400
|
|
|
|
|
|
def test_mass_assignment_ignored(env):
|
|
"""Лишние поля (is_admin, password_hash) при смене профиля техником не применяются — PATCH /users ему недоступен."""
|
|
c, t = env["c"], env["h"]["sec_a"]
|
|
r = c.patch(f"{V1}/users/{env['ids']['sec_a']}", headers=t, json={"is_admin": True})
|
|
assert r.status_code == 403
|
|
assert c.get(f"{V1}/me", headers=t).json()["is_admin"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------- изоляция данных (IDOR)
|
|
def test_technician_sees_only_own_orders(env):
|
|
c, h, oid = env["c"], env["h"], env["order"]["id"]
|
|
assert c.get(f"{V1}/orders/{oid}", headers=h["sec_a"]).status_code == 200
|
|
assert c.get(f"{V1}/orders/{oid}", headers=h["sec_b"]).status_code == 404
|
|
assert oid not in [o["id"] for o in c.get(f"{V1}/orders", headers=h["sec_b"]).json()]
|
|
# поиск и счётчики тоже не раскрывают чужие наряды
|
|
assert c.get(f"{V1}/orders", headers=h["sec_b"], params={"q": "sec-order"}).json() == []
|
|
assert c.get(f"{V1}/orders/{oid}/comments", headers=h["sec_b"]).status_code == 404
|
|
|
|
|
|
def test_foreign_order_is_untouchable(env):
|
|
c, b, oid = env["c"], env["h"]["sec_b"], env["order"]["id"]
|
|
item = env["order"]["items"][0]["id"]
|
|
for m, p, body in (("POST", f"/orders/{oid}/start", None),
|
|
("PUT", f"/orders/{oid}/items/{item}", {"value": "ok"}),
|
|
("POST", f"/orders/{oid}/submit", None),
|
|
("POST", f"/orders/{oid}/review", {"approve": True, "comment": ""}),
|
|
("POST", f"/orders/{oid}/sync", {"items": [{"item_id": item, "value": "ok"}], "submit": True}),
|
|
("POST", f"/orders/{oid}/comments", {"text": "hi"})):
|
|
r = c.request(m, V1 + p, headers=b, json=body)
|
|
assert r.status_code in (403, 404), f"{m} {p} -> {r.status_code}"
|
|
r = c.post(f"{V1}/orders/{oid}/items/{item}/photos", headers={**b, "Content-Type": "image/png"}, content=PNG)
|
|
assert r.status_code in (403, 404)
|
|
|
|
|
|
def test_photo_access_is_scoped_to_order(env):
|
|
c, h, o = env["c"], env["h"], env["order"]
|
|
item = o["items"][0]["id"]
|
|
r = c.post(f"{V1}/orders/{o['id']}/items/{item}/photos", headers={**h["sec_a"], "Content-Type": "image/png"}, content=PNG)
|
|
assert r.status_code == 201, r.text
|
|
pid = r.json()["items"][0]["photos"][0]["id"]
|
|
assert c.get(f"{V1}/photos/{pid}", headers=h["sec_a"]).status_code == 200
|
|
assert c.get(f"{V1}/photos/{pid}", headers=h["sec_eng"]).status_code == 200
|
|
assert c.get(f"{V1}/photos/{pid}", headers=h["sec_b"]).status_code == 404 # чужой наряд
|
|
assert c.get(f"{V1}/photos/{pid}").status_code == 401 # без токена
|
|
assert c.delete(f"{V1}/photos/{pid}", headers=h["sec_b"]).status_code == 404
|
|
|
|
|
|
def test_cannot_answer_item_of_another_order(env):
|
|
"""item_id из другого наряда не принимается в рамках своего."""
|
|
c, h = env["c"], env["h"]
|
|
o2 = c.post(f"{V1}/orders", headers=h["sec_eng"], json={"title": "sec-o2", "checklist_id": env["cl"]["id"],
|
|
"assignee_user_id": env["ids"]["sec_a"]}).json()
|
|
foreign_item = o2["items"][0]["id"]
|
|
r = c.put(f"{V1}/orders/{env['order']['id']}/items/{foreign_item}", headers=h["sec_a"], json={"value": "ok"})
|
|
assert r.status_code == 404
|
|
r = c.post(f"{V1}/orders/{env['order']['id']}/sync", headers=h["sec_a"], json={"items": [{"item_id": foreign_item, "value": "ok"}]})
|
|
assert r.json()["results"][0]["status"] == "error"
|
|
|
|
|
|
def test_comment_delete_only_own(env):
|
|
c, h, oid = env["c"], env["h"], env["order"]["id"]
|
|
cid = c.post(f"{V1}/orders/{oid}/comments", headers=h["sec_a"], json={"text": "мой"}).json()["id"]
|
|
assert c.delete(f"{V1}/comments/{cid}", headers=h["sec_b"]).status_code == 403
|
|
assert c.delete(f"{V1}/comments/{cid}", headers=h["sec_a"]).status_code == 204
|
|
|
|
|
|
def test_executor_cannot_review_own_order_and_state_machine(env):
|
|
c, h = env["c"], env["h"]
|
|
o = c.post(f"{V1}/orders", headers=h["sec_eng"], json={"title": "sec-sm", "checklist_id": env["cl"]["id"],
|
|
"assignee_user_id": env["ids"]["sec_a"], "reviewer_user_id": env["ids"]["sec_eng"]}).json()
|
|
oid, items = o["id"], o["items"]
|
|
assert c.post(f"{V1}/orders/{oid}/review", headers=h["sec_a"], json={"approve": True, "comment": ""}).status_code == 403 # не на проверке
|
|
c.put(f"{V1}/orders/{oid}/items/{items[0]['id']}", headers=h["sec_a"], json={"value": "done"})
|
|
c.put(f"{V1}/orders/{oid}/items/{items[1]['id']}", headers=h["sec_a"], json={"value": "3"})
|
|
assert c.post(f"{V1}/orders/{oid}/submit", headers=h["sec_a"]).json()["status"] == "review"
|
|
assert c.post(f"{V1}/orders/{oid}/review", headers=h["sec_a"], json={"approve": True, "comment": ""}).status_code == 403
|
|
assert c.put(f"{V1}/orders/{oid}/items/{items[1]['id']}", headers=h["sec_a"], json={"value": "4"}).status_code == 403 # после сдачи нельзя править
|
|
assert c.post(f"{V1}/orders/{oid}/review", headers=h["sec_eng"], json={"approve": True, "comment": ""}).json()["status"] == "completed"
|
|
assert c.post(f"{V1}/orders/{oid}/cancel", headers=h["sec_eng"]).status_code == 400 # завершённый не отменить
|
|
|
|
|
|
def test_answer_validation_rejects_garbage(env):
|
|
c, h = env["c"], env["h"]
|
|
o = c.post(f"{V1}/orders", headers=h["sec_eng"], json={"title": "sec-val", "checklist_id": env["cl"]["id"],
|
|
"assignee_user_id": env["ids"]["sec_a"]}).json()
|
|
num = o["items"][1]["id"]
|
|
for bad in ("abc", "1e999999", "NaN", "<script>"):
|
|
r = c.put(f"{V1}/orders/{o['id']}/items/{num}", headers=h["sec_a"], json={"value": bad})
|
|
assert r.status_code in (422,), f"{bad!r} -> {r.status_code}"
|
|
|
|
|
|
# ---------------------------------------------------------------- загрузка файлов
|
|
def test_upload_rejects_non_images(env):
|
|
c, h, o = env["c"], env["h"], env["order"]
|
|
item = o["items"][0]["id"]
|
|
url = f"{V1}/orders/{o['id']}/items/{item}/photos"
|
|
payloads = {
|
|
"html": b"<html><script>alert(1)</script></html>",
|
|
"svg": b'<svg xmlns="http://www.w3.org/2000/svg" onload="alert(1)"/>',
|
|
"exe": b"MZ\x90\x00" + b"\0" * 100,
|
|
"php": b"<?php system($_GET['c']); ?>",
|
|
"empty": b"",
|
|
}
|
|
for name, data in payloads.items():
|
|
r = c.post(url, headers={**h["sec_a"], "Content-Type": "image/png"}, content=data) # Content-Type подделан
|
|
assert r.status_code == 415, f"{name} -> {r.status_code}"
|
|
|
|
|
|
def test_upload_size_limit(env):
|
|
c, h, o = env["c"], env["h"], env["order"]
|
|
c.put(f"{V1}/settings", headers=h["admin"], json={"max_photo_mb": 1})
|
|
try:
|
|
big = PNG + b"\0" * (1024 * 1024 + 10)
|
|
r = c.post(f"{V1}/orders/{o['id']}/items/{o['items'][0]['id']}/photos",
|
|
headers={**h["sec_a"], "Content-Type": "image/png"}, content=big)
|
|
assert r.status_code == 413
|
|
finally:
|
|
c.put(f"{V1}/settings", headers=h["admin"], json={"max_photo_mb": 10})
|
|
|
|
|
|
def test_stored_photo_name_is_server_generated(env):
|
|
"""Имя файла на диске не зависит от клиента (нет path traversal через Idempotency-Key/заголовки)."""
|
|
c, h, o = env["c"], env["h"], env["order"]
|
|
r = c.post(f"{V1}/orders/{o['id']}/items/{o['items'][0]['id']}/photos",
|
|
headers={**h["sec_a"], "Content-Type": "image/png", "Idempotency-Key": "../../../../etc/passwd"}, content=PNG + b"x")
|
|
assert r.status_code == 201
|
|
from app.db import SessionLocal
|
|
from app.models import Photo
|
|
with SessionLocal() as s:
|
|
for p in s.query(Photo).all():
|
|
parts = p.path.split("/")
|
|
parts = parts[1:] if parts[0] == "keep" else parts # keep/ — фото «хранить более года»
|
|
assert ".." not in p.path and not p.path.startswith(("/", "\\")) and len(parts) == 3
|
|
|
|
|
|
def test_static_path_traversal_blocked(env):
|
|
c = env["c"]
|
|
for p in ("/static/../app/config.py", "/static/%2e%2e/app/config.py", "/static/..%2fapp%2fconfig.py",
|
|
"/static/m/../../main.py", "/static/%252e%252e/app/main.py"):
|
|
r = c.get(p)
|
|
assert r.status_code in (400, 404) or "SECRET_KEY" not in r.text, p
|
|
assert "SECRET_KEY" not in r.text
|
|
|
|
|
|
# ---------------------------------------------------------------- инъекции
|
|
def test_sql_injection_in_login_and_search(env):
|
|
c, h = env["c"], env["h"]
|
|
for u in ("admin'--", "' OR '1'='1", "admin\" OR \"1\"=\"1", "'; DROP TABLE users;--"):
|
|
assert c.post(f"{V1}/auth/login", json={"username": u, "password": u}).status_code == 401
|
|
catalog._fails.clear()
|
|
for q in ("'; DROP TABLE orders;--", "' OR 1=1--", "%", "_", "\\"):
|
|
r = c.get(f"{V1}/orders", headers=h["admin"], params={"q": q})
|
|
assert r.status_code == 200
|
|
if q in ("%", "_", "\\"): # шаблонные символы LIKE экранируются: «%» ищется как обычный символ, а не «всё подряд»
|
|
assert all(q in o["title"] for o in r.json()), q
|
|
assert c.get(f"{V1}/orders", headers=h["admin"]).json() # таблица на месте
|
|
assert c.get(f"{V1}/orders", headers=h["admin"], params={"status": "new' OR '1'='1"}).json() == []
|
|
|
|
|
|
def test_xss_payload_is_stored_as_text_and_served_as_json(env):
|
|
c, h, oid = env["c"], env["h"], env["order"]["id"]
|
|
payload = "<img src=x onerror=alert(1)>"
|
|
r = c.post(f"{V1}/orders/{oid}/comments", headers=h["sec_a"], json={"text": payload})
|
|
assert r.status_code == 201 and r.headers["content-type"].startswith("application/json")
|
|
lst = c.get(f"{V1}/orders/{oid}/comments", headers=h["sec_a"])
|
|
assert lst.headers["content-type"].startswith("application/json")
|
|
assert payload in lst.text
|
|
|
|
|
|
def test_frontend_never_uses_innerhtml():
|
|
"""Интерфейс строит DOM через h()/textContent: опасных приёмников XSS нет."""
|
|
from pathlib import Path
|
|
root = Path(__file__).resolve().parents[1] / "app" / "static"
|
|
bad = ("innerHTML", "outerHTML", "insertAdjacentHTML", "document.write", "eval(", "new Function(")
|
|
for f in list(root.glob("*.js")) + list((root / "m").glob("*.js")):
|
|
src = f.read_text(encoding="utf-8")
|
|
for token in bad:
|
|
if token in src:
|
|
# допустимо только присваивание пустой строки (очистка контейнера)
|
|
import re
|
|
for m in re.finditer(re.escape(token) + r"\s*=\s*([^;\n]+)", src):
|
|
assert m.group(1).strip() in ('""', "''"), f"{f.name}: {m.group(0)}"
|
|
if token not in ("innerHTML",):
|
|
raise AssertionError(f"{f.name}: {token}")
|
|
|
|
|
|
def test_report_params_are_validated(env):
|
|
c, e = env["c"], env["h"]["sec_eng"]
|
|
base = f"{V1}/reports/completed-orders"
|
|
assert c.get(base, headers=e, params={"from": "2026-02-01T00:00", "to": "2026-01-01T00:00"}).status_code == 422
|
|
assert c.get(base, headers=e, params={"from": "2020-01-01T00:00", "to": "2026-01-01T00:00"}).status_code == 422 # > 366 дней
|
|
assert c.get(base, headers=e, params={"from": "x", "to": "y"}).status_code == 422
|
|
assert c.get(base, headers=e, params={"from": "2026-01-01T00:00", "to": "2026-02-01T00:00", "format": "../x"}).status_code == 422
|
|
|
|
|
|
# ---------------------------------------------------------------- пароли, push, лимиты, секреты
|
|
def test_password_policy(env):
|
|
c, a = env["c"], env["h"]["admin"]
|
|
for pw in ("1234", "123456789", "x" * 73, "я" * 40): # короткий, 9 символов, >72 байт, >72 байт в UTF-8
|
|
r = c.post(f"{V1}/users", headers=a, json={"username": "sec_weak", "password": pw})
|
|
assert r.status_code == 422, pw
|
|
assert c.post(f"{V1}/users", headers=a, json={"username": "sec_ok", "password": "x" * 72}).status_code == 201
|
|
uid = env["ids"]["sec_a"]
|
|
assert c.patch(f"{V1}/users/{uid}", headers=a, json={"password": "short"}).status_code == 422
|
|
r = c.post(f"{V1}/auth/change-password", headers=env["h"]["sec_b"], json={"old_password": "Passw0rd!x", "new_password": "short"})
|
|
assert r.status_code == 422
|
|
|
|
|
|
def test_long_password_does_not_crash(env):
|
|
r = env["c"].post(f"{V1}/users", headers=env["h"]["admin"], json={"username": "sec_long", "password": "A" * 100})
|
|
assert r.status_code == 422 # раньше bcrypt молча обрезал до 72 байт
|
|
|
|
|
|
def test_push_endpoint_is_restricted(env):
|
|
c, h, keys = env["c"], env["h"]["sec_a"], {"p256dh": "x" * 20, "auth": "y" * 10}
|
|
bad = ["http://127.0.0.1:8000/api/v1/health", "https://127.0.0.1/x", "https://localhost/x", "https://[::1]/x",
|
|
"https://192.168.0.3/x", "https://169.254.169.254/latest/meta-data", "http://fcm.googleapis.com/fcm/send/x",
|
|
"https://evil.example.com/x", "https://fcm.googleapis.com.evil.com/x", "https://evilgoogleapis.com/x",
|
|
"https://user:pw@fcm.googleapis.com/x", "https://fcm.googleapis.com:8443/x", "file:///etc/passwd", "gopher://x", ""]
|
|
for ep in bad:
|
|
r = c.post(f"{V1}/push/subscribe", headers=h, json={"endpoint": ep, "keys": keys})
|
|
assert r.status_code == 422, ep
|
|
good = ["https://fcm.googleapis.com/fcm/send/abc", "https://updates.push.services.mozilla.com/wpush/v2/abc",
|
|
"https://web.push.apple.com/abc", "https://wns2-par02p.notify.windows.com/?token=abc"]
|
|
for ep in good:
|
|
assert c.post(f"{V1}/push/subscribe", headers=h, json={"endpoint": ep, "keys": keys}).status_code == 204, ep
|
|
assert c.post(f"{V1}/push/unsubscribe", headers=h, json={"endpoint": ep, "keys": {}}).status_code == 204
|
|
# без ключей подписка не принимается (раньше — 500)
|
|
r = c.post(f"{V1}/push/subscribe", headers=h, json={"endpoint": good[0], "keys": {}})
|
|
assert r.status_code == 422
|
|
|
|
|
|
def test_login_lockout_is_per_address_and_login(env):
|
|
"""Посторонний не может заблокировать вход настоящему пользователю (другой адрес), но перебор ограничен."""
|
|
victim, other = env["c"], TestClient(app, client=("203.0.113.9", 4000))
|
|
body = {"username": "sec_b", "password": "wrong-password"}
|
|
codes = [other.post(f"{V1}/auth/login", json=body).status_code for _ in range(config.LOGIN_MAX_FAILS + 1)]
|
|
assert codes[-1] == 429 # злоумышленник заблокирован…
|
|
ok = victim.post(f"{V1}/auth/login", json={"username": "sec_b", "password": "Passw0rd!x"})
|
|
assert ok.status_code == 200 # …а владелец учётной записи входит
|
|
# перебор по разным логинам с одного адреса упирается в общий предел адреса
|
|
catalog._fails.clear()
|
|
n = config.LOGIN_MAX_FAILS * 6
|
|
codes = [other.post(f"{V1}/auth/login", json={"username": f"user{i}", "password": "x"}).status_code for i in range(n + 2)]
|
|
assert 429 in codes and codes.index(429) == n
|
|
# распределённый перебор одного логина с разных адресов упирается в общий предел логина
|
|
catalog._fails.clear()
|
|
codes = []
|
|
for i in range(config.LOGIN_MAX_FAILS * 4 + 1):
|
|
cl = TestClient(app, client=(f"198.51.100.{i + 1}", 5000))
|
|
codes.append(cl.post(f"{V1}/auth/login", json={"username": "sec_b", "password": "bad"}).status_code)
|
|
assert codes[-1] == 429
|
|
|
|
|
|
def test_rate_limiter_memory_is_bounded():
|
|
catalog._fails.clear()
|
|
old = time.time() - config.LOGIN_WINDOW_SEC - 10
|
|
for i in range(catalog.MAX_TRACKED + 500):
|
|
catalog._fails[f"ipuser:x{i}|u"] = [old]
|
|
catalog._check_rate("ip:probe", 5)
|
|
assert len(catalog._fails) < 10
|
|
now = time.time()
|
|
for i in range(catalog.MAX_TRACKED + 500): # свежие записи не истекли — предел всё равно действует
|
|
catalog._fails[f"k{i}"] = [now]
|
|
catalog._check_rate("ip:probe", 5)
|
|
assert len(catalog._fails) <= catalog.MAX_TRACKED
|
|
catalog._fails.clear()
|
|
|
|
|
|
def test_settings_show_client_address_for_proxy_check(env):
|
|
r = env["c"].get(f"{V1}/settings", headers={**env["h"]["admin"], "X-Forwarded-For": "1.2.3.4"})
|
|
assert r.status_code == 200 and "client_ip" in r.json() and r.json()["forwarded_for"] == "1.2.3.4"
|
|
|
|
|
|
def test_weak_secrets_refuse_to_start_in_production(monkeypatch):
|
|
from app import main
|
|
monkeypatch.setattr(config, "PRODUCTION", True)
|
|
for key in ("", "dev-secret-change-me", "short-key", "a" * 31):
|
|
monkeypatch.setattr(config, "SECRET_KEY", key)
|
|
with pytest.raises(RuntimeError):
|
|
main.check_secrets()
|
|
monkeypatch.setattr(config, "SECRET_KEY", "k" * 40)
|
|
main.check_secrets() # нормальный ключ — запуск разрешён
|
|
monkeypatch.setattr(config, "ADMIN_PASSWORD", "admin")
|
|
main.check_secrets(users_exist=True) # пароль админа проверяется только при первом создании
|
|
with pytest.raises(RuntimeError):
|
|
main.check_secrets(users_exist=False)
|
|
monkeypatch.setattr(config, "PRODUCTION", False)
|
|
monkeypatch.setattr(config, "SECRET_KEY", "")
|
|
main.check_secrets() # в разработке — только предупреждение
|
|
|
|
|
|
def test_deploy_files_do_not_trust_all_proxies():
|
|
from pathlib import Path
|
|
root = Path(__file__).resolve().parents[1]
|
|
docker = (root / "Dockerfile").read_text(encoding="utf-8")
|
|
compose = (root / "docker-compose.yml").read_text(encoding="utf-8")
|
|
assert "forwarded-allow-ips \\\"*\\\"" not in docker and "FORWARDED_ALLOW_IPS" in docker
|
|
assert "FORWARDED_ALLOW_IPS" in compose and "APP_ENV: production" in compose
|
|
|
|
|
|
def test_android_forbids_cleartext_in_release():
|
|
from pathlib import Path
|
|
android = Path(__file__).resolve().parents[1] / "android" / "app" / "src"
|
|
manifest = (android / "main" / "AndroidManifest.xml").read_text(encoding="utf-8")
|
|
assert 'usesCleartextTraffic="true"' not in manifest and "networkSecurityConfig" in manifest
|
|
cfg = (android / "main" / "res" / "xml" / "network_security_config.xml").read_text(encoding="utf-8")
|
|
assert 'cleartextTrafficPermitted="false"' in cfg and 'cleartextTrafficPermitted="true"' not in cfg
|
|
|
|
|
|
# ---------------------------------------------------------------- известные недостатки (xfail)
|
|
@pytest.mark.xfail(reason="SEC-05: нет заголовков безопасности (nosniff, X-Frame-Options, CSP)", strict=True)
|
|
def test_gap_security_headers(env):
|
|
r = env["c"].get("/")
|
|
assert r.headers.get("x-content-type-options") == "nosniff"
|
|
assert r.headers.get("x-frame-options") or "frame-ancestors" in r.headers.get("content-security-policy", "")
|
|
|
|
|
|
def test_csv_formula_injection_is_escaped(env):
|
|
from app.reports import to_csv
|
|
row = {"order_id": 1, "finished_at": None, "equipment": "=HYPERLINK(\"http://evil\")", "inventory_no": "", "location": "",
|
|
"checklist": "+cmd", "executor": "@x", "position": "-1", "status": "completed", "deviations": 0}
|
|
cells = to_csv([row]).decode("utf-8-sig").splitlines()[1].split(";")
|
|
assert not any(x.lstrip('"').startswith(("=", "+", "-", "@")) for x in cells[2:8])
|
|
assert cells[2].lstrip('"').startswith("'=HYPERLINK")
|
|
|
|
|
|
@pytest.mark.xfail(reason="SEC-07: документация API (/docs, /openapi.json) доступна без авторизации", strict=True)
|
|
def test_gap_docs_public(env):
|
|
assert env["c"].get("/openapi.json").status_code in (401, 403, 404)
|