347 lines
14 KiB
Python
347 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""fuzz_test.py — локальные злонамеренные и экстремальные сценарии по стендам."""
|
||
|
||
if __name__ != "__main__":
|
||
import pytest
|
||
|
||
pytest.skip("standalone local fuzz scenarios; run explicitly with python", allow_module_level=True)
|
||
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "site"))
|
||
|
||
from app import app
|
||
import json
|
||
import config.loader as config
|
||
|
||
client = app.test_client()
|
||
H = {"Content-Type": "application/json"}
|
||
AUTH = os.getenv("MOCK_AUTH_TOKEN", "")
|
||
MH = {**H, **({"X-Mock-Auth": AUTH} if AUTH else {})}
|
||
stands = config.list_stands()
|
||
passed = 0
|
||
failed = 0
|
||
errors = []
|
||
|
||
def t(name, fn):
|
||
global passed, failed
|
||
try:
|
||
fn()
|
||
passed += 1
|
||
print(f" ✅ {name}")
|
||
except AssertionError as e:
|
||
failed += 1
|
||
msg = f" ❌ {name}: {e}"
|
||
print(msg)
|
||
errors.append(msg)
|
||
except Exception as e:
|
||
failed += 1
|
||
msg = f" 💥 {name}: {type(e).__name__}: {e}"
|
||
print(msg)
|
||
errors.append(msg)
|
||
|
||
def st(r, exp, note=""):
|
||
extra = f" ({note})" if note else ""
|
||
assert r.status_code == exp, f"expected {exp}, got {r.status_code}: {r.data[:200]!r}{extra}"
|
||
|
||
def js(r):
|
||
try:
|
||
return r.get_json()
|
||
except:
|
||
raise AssertionError(f"not JSON: {r.data[:200]!r}")
|
||
|
||
|
||
for sid in stands:
|
||
base = f"/{sid}"
|
||
print(f"\n{'='*55}")
|
||
print(f" {sid.upper()} — фаззинг")
|
||
print(f"{'='*55}")
|
||
|
||
# Reset state
|
||
client.post(f"{base}/api/v1/svc/_mock/reset", headers=MH)
|
||
|
||
# ═══ 1. MALICIOUS JSON ═══
|
||
print("\n── Злонамеренный JSON ──")
|
||
|
||
t("SQL injection", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": "'; DROP TABLE instances; --"}, headers=H), 201))
|
||
|
||
t("XSS <script>", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": "<script>alert(1)</script>"}, headers=H), 201))
|
||
|
||
t("Unicode-emoji", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": "🔥💀🎉日本語한국어"}, headers=H), 201))
|
||
|
||
t("Null-байт", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": "test\x00null"}, headers=H), 201))
|
||
|
||
t("10000 символов", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": "A" * 10000}, headers=H), 201))
|
||
|
||
t("Пустой body", lambda: st(client.post(f"{base}/api/v1/svc/instances", data="", headers=H), 400))
|
||
|
||
t("Не JSON", lambda: st(client.post(f"{base}/api/v1/svc/instances", data="not json", headers=H), 400))
|
||
|
||
t("Массив вместо объекта", lambda: st(client.post(f"{base}/api/v1/svc/instances", json=[1,2,3], headers=H), 400))
|
||
|
||
# ═══ 2. WRONG TYPES ═══
|
||
print("\n── Неверные типы ──")
|
||
|
||
t("serviceId='abc'", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": "abc", "displayName": "x"}, headers=H), 400))
|
||
|
||
t("serviceId=1.5", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1.5, "displayName": "x"}, headers=H), 400))
|
||
|
||
t("serviceId=-1", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": -1, "displayName": "x"}, headers=H), 400))
|
||
|
||
t("serviceId=0", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 0, "displayName": "x"}, headers=H), 400))
|
||
|
||
t("displayName=число", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": 12345}, headers=H), 400))
|
||
|
||
t("paramId строка", lambda: st(client.post(f"{base}/api/v1/svc/instanceOperationCfsParams",
|
||
json={"instanceOperationUid": "00000000-0000-0000-0000-000000000000",
|
||
"svcOperationCfsParamId": "abc", "paramValue": "x"}, headers=H), 400))
|
||
|
||
# ═══ 3. PAGINATION ═══
|
||
print("\n── Пагинация ──")
|
||
|
||
def test_page_size_upper_bound():
|
||
d = js(client.get(f"{base}/api/v1/svc/instances?pageSize=999999"))
|
||
assert d["pageSize"] == 200
|
||
t("pageSize=999999", test_page_size_upper_bound)
|
||
|
||
t("pageSize=0", lambda: st(client.get(f"{base}/api/v1/svc/instances?pageSize=0"), 200))
|
||
|
||
t("pageSize=-5", lambda: st(client.get(f"{base}/api/v1/svc/instances?pageSize=-5"), 200))
|
||
|
||
t("pageSize=''", lambda: st(client.get(f"{base}/api/v1/svc/instances?pageSize="), 200))
|
||
|
||
def test_page_lower_bound():
|
||
d = js(client.get(f"{base}/api/v1/svc/instances?page=-100"))
|
||
assert d["page"] == 1
|
||
t("page=-100", test_page_lower_bound)
|
||
|
||
# ═══ 4. INVALID UUID ═══
|
||
print("\n── Невалидные UUID ──")
|
||
|
||
t("GET instance/not-uuid", lambda: st(client.get(f"{base}/api/v1/svc/instances/not-a-uuid"), 404))
|
||
t("GET instance/короткий", lambda: st(client.get(f"{base}/api/v1/svc/instances/abc"), 404))
|
||
t("GET op/not-uuid", lambda: st(client.get(f"{base}/api/v1/svc/instanceOperations/not-a-uuid"), 404))
|
||
t("POST run/not-uuid", lambda: st(client.post(f"{base}/api/v1/svc/instanceOperations/not-a-uuid/run", headers=H), 404))
|
||
|
||
# ═══ 5. HTTP METHOD ABUSE ═══
|
||
print("\n── HTTP-методы ──")
|
||
|
||
t("PUT /services", lambda: st(client.put(f"{base}/api/v1/svc/services"), 405))
|
||
t("DELETE /services", lambda: st(client.delete(f"{base}/api/v1/svc/services"), 405))
|
||
t("PATCH /run", lambda: st(client.patch(f"{base}/api/v1/svc/instanceOperations/x/run"), 405))
|
||
|
||
# ═══ 6. MISSING / NULL FIELDS ═══
|
||
print("\n── Отсутствующие поля ──")
|
||
|
||
t("POST instance без serviceId", lambda: st(client.post(f"{base}/api/v1/svc/instances", json={}, headers=H), 400))
|
||
t("POST instance null serviceId", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": None, "displayName": "x"}, headers=H), 400))
|
||
t("POST op без instanceUid", lambda: st(client.post(f"{base}/api/v1/svc/instanceOperations",
|
||
json={"operation": "create"}, headers=H), 400))
|
||
t("POST cfsParams без paramId", lambda: st(client.post(f"{base}/api/v1/svc/instanceOperationCfsParams",
|
||
json={"instanceOperationUid": "00000000-0000-0000-0000-000000000000"}, headers=H), 400))
|
||
|
||
# ═══ 7. SPECIAL CHARS ═══
|
||
print("\n── Спецсимволы ──")
|
||
|
||
def test_special_chars():
|
||
r = client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": "special"}, headers=H)
|
||
iuid = js(r)["instanceUid"]
|
||
r2 = client.post(f"{base}/api/v1/svc/instanceOperations",
|
||
json={"instanceUid": iuid, "operation": "create", "svcOperationId": 18}, headers=H)
|
||
opid = js(r2)["instanceOperationUid"]
|
||
st(client.post(f"{base}/api/v1/svc/instanceOperationCfsParams",
|
||
json={"instanceOperationUid": opid, "svcOperationCfsParamId": 242,
|
||
"paramValue": "line1\nline2\rline3\tline4"}, headers=H), 200)
|
||
t("paramValue \\n\\r\\t", test_special_chars)
|
||
|
||
def test_empty_param():
|
||
r = client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": "empty-param"}, headers=H)
|
||
iuid = js(r)["instanceUid"]
|
||
r2 = client.post(f"{base}/api/v1/svc/instanceOperations",
|
||
json={"instanceUid": iuid, "operation": "create", "svcOperationId": 18}, headers=H)
|
||
opid = js(r2)["instanceOperationUid"]
|
||
st(client.post(f"{base}/api/v1/svc/instanceOperationCfsParams",
|
||
json={"instanceOperationUid": opid, "svcOperationCfsParamId": 242,
|
||
"paramValue": ""}, headers=H), 200)
|
||
# Run with empty params — should work
|
||
d = js(client.post(f"{base}/api/v1/svc/instanceOperations/{opid}/run", headers=H))
|
||
assert d["ok"] is True
|
||
t("paramValue='' + run", test_empty_param)
|
||
|
||
t("displayName JSON-like", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": '{"key":"val"}'}, headers=H), 201))
|
||
|
||
# ═══ 8. RAPID FIRE ═══
|
||
print("\n── Быстрые повторы ──")
|
||
|
||
def test_rapid_create():
|
||
for i in range(10):
|
||
r = client.post(f"{base}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": f"rapid-{i}"}, headers=H)
|
||
assert r.status_code == 201, f"rapid {i}: {r.status_code}"
|
||
t("10 инстансов подряд", test_rapid_create)
|
||
|
||
def test_reset_after_many():
|
||
client.post(f"{base}/api/v1/svc/_mock/reset", headers=MH)
|
||
d = js(client.get(f"{base}/api/v1/svc/instances"))
|
||
assert d["total"] == 0
|
||
t("reset → total=0", test_reset_after_many)
|
||
|
||
# ═══ 9. PATH TRAVERSAL ═══
|
||
print("\n── Path traversal ──")
|
||
|
||
t("GET /../services → 404", lambda: st(client.get(f"{base}/api/v1/svc/../services"), 404))
|
||
|
||
def test_double_slash_is_normalized():
|
||
response = client.get(f"{base}//api/v1/svc/services")
|
||
st(response, 200)
|
||
assert "results" in js(response)
|
||
t("GET двойной слеш → нормализованный список", test_double_slash_is_normalized)
|
||
|
||
# ═══ 10. DELAY ═══
|
||
print("\n── Задержка ──")
|
||
|
||
t("delay 999", lambda: st(client.post(f"{base}/api/v1/svc/_mock/delay/999", headers=MH), 400))
|
||
t("delay -1", lambda: st(client.post(f"{base}/api/v1/svc/_mock/delay/-1", headers=MH), 400))
|
||
def test_zero_delay():
|
||
d = js(client.post(f"{base}/api/v1/svc/_mock/delay/0", headers=MH))
|
||
assert d["delay"] == 0
|
||
t("delay 0", test_zero_delay)
|
||
|
||
def test_max_delay():
|
||
d = js(client.post(f"{base}/api/v1/svc/_mock/delay/5", headers=MH))
|
||
assert d["delay"] == 5.0
|
||
t("delay 5.0", test_max_delay)
|
||
client.post(f"{base}/api/v1/svc/_mock/delay/0.1", headers=MH)
|
||
|
||
# ═══ 11. FIELDS ═══
|
||
print("\n── Параметр fields ──")
|
||
|
||
t("fields=nonexistent", lambda: st(client.get(f"{base}/api/v1/svc/instances?fields=nonexistent"), 200))
|
||
t("fields=*", lambda: st(client.get(f"{base}/api/v1/svc/instances?fields=*"), 200))
|
||
|
||
# ═══ 12. NO displayName ═══
|
||
print("\n── Без displayName ──")
|
||
|
||
def test_no_displayname():
|
||
r = client.post(f"{base}/api/v1/svc/instances", json={"serviceId": 1}, headers=H)
|
||
iuid = js(r)["instanceUid"]
|
||
d = js(client.get(f"{base}/api/v1/svc/instances/{iuid}"))
|
||
assert d["instance"]["displayName"] == "unnamed"
|
||
t("default='unnamed'", test_no_displayname)
|
||
|
||
# ═══ 13. PAGES ═══
|
||
print("\n── Страницы ──")
|
||
|
||
t("GET /{sid}/ → HTML", lambda: st(client.get(f"{base}/"), 200))
|
||
|
||
t("GET /swagger → HTML", lambda: st(client.get("/swagger"), 200))
|
||
|
||
t("GET /health → OK", lambda: (
|
||
r := client.get("/health"), st(r, 200), r.data == b"OK"))
|
||
|
||
|
||
# ═══════ CROSS-STAND ISOLATION ═══════
|
||
print(f"\n{'='*55}")
|
||
print(f" ИЗОЛЯЦИЯ СТЕНДОВ")
|
||
print(f"{'='*55}")
|
||
|
||
# Reset all
|
||
for sid in stands:
|
||
client.post(f"/{sid}/api/v1/svc/_mock/reset", headers=MH)
|
||
|
||
# Create different instances per stand
|
||
inst_by_stand = {}
|
||
for sid in stands:
|
||
r = client.post(f"/{sid}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": f"only-on-{sid}"}, headers=H)
|
||
inst_by_stand[sid] = js(r)["instanceUid"]
|
||
|
||
for sid in stands:
|
||
def check_isolated(sid=sid):
|
||
d = js(client.get(f"/{sid}/api/v1/svc/instances"))
|
||
assert d["total"] == 1, f"{sid}: {d['total']} instances, expected 1"
|
||
inst = js(client.get(f"/{sid}/api/v1/svc/instances/{inst_by_stand[sid]}"))
|
||
assert inst["instance"]["displayName"] == f"only-on-{sid}"
|
||
t(f"{sid}: только свой инстанс", check_isolated)
|
||
|
||
# Cross-stand checks apply only when at least two stands are configured.
|
||
if len(stands) >= 2:
|
||
first_stand, second_stand = stands[:2]
|
||
t(
|
||
"чужая instance → 404",
|
||
lambda: st(
|
||
client.get(
|
||
f"/{first_stand}/api/v1/svc/instances/{inst_by_stand[second_stand]}"
|
||
),
|
||
404,
|
||
),
|
||
)
|
||
|
||
client.post(f"/{first_stand}/api/v1/svc/_mock/reset", headers=MH)
|
||
client.post(f"/{second_stand}/api/v1/svc/_mock/reset", headers=MH)
|
||
client.post(f"/{first_stand}/api/v1/svc/_mock/fail-next", headers=MH)
|
||
|
||
def test_fail_isolation():
|
||
first_instance = js(client.post(
|
||
f"/{first_stand}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": "fail-isolation"}, headers=H,
|
||
))["instanceUid"]
|
||
first_operation = js(client.post(
|
||
f"/{first_stand}/api/v1/svc/instanceOperations",
|
||
json={"instanceUid": first_instance, "operation": "create", "svcOperationId": 18},
|
||
headers=H,
|
||
))["instanceOperationUid"]
|
||
first_result = js(client.post(
|
||
f"/{first_stand}/api/v1/svc/instanceOperations/{first_operation}/run",
|
||
headers=H,
|
||
))
|
||
assert first_result["ok"] is False
|
||
|
||
second_instance = js(client.post(
|
||
f"/{second_stand}/api/v1/svc/instances",
|
||
json={"serviceId": 1, "displayName": "normal"}, headers=H,
|
||
))["instanceUid"]
|
||
second_operation = js(client.post(
|
||
f"/{second_stand}/api/v1/svc/instanceOperations",
|
||
json={"instanceUid": second_instance, "operation": "create", "svcOperationId": 18},
|
||
headers=H,
|
||
))["instanceOperationUid"]
|
||
second_result = js(client.post(
|
||
f"/{second_stand}/api/v1/svc/instanceOperations/{second_operation}/run",
|
||
headers=H,
|
||
))
|
||
assert second_result["ok"] is True
|
||
|
||
t("fail-next изоляция между стендами", test_fail_isolation)
|
||
else:
|
||
print(" Cross-stand isolation skipped: fewer than two stands configured")
|
||
|
||
|
||
# ═══════ SUMMARY ═══════
|
||
print(f"\n{'='*55}")
|
||
total = passed + failed
|
||
print(f" PASS: {passed}/{total}")
|
||
if failed:
|
||
print(f" FAIL: {failed}")
|
||
for e in errors:
|
||
print(f" {e}")
|
||
else:
|
||
print(f" 🎉 ВСЕ ТЕСТЫ ПРОЙДЕНЫ — {total} проверок, {len(stands)} стенд(а)")
|
||
print(f"{'='*55}")
|