308 lines
13 KiB
Python
308 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""fuzz_test.py — фаззинг Polygon: злонамеренные и экстремальные сценарии × 3 стенда."""
|
|
|
|
from app import app
|
|
import json
|
|
|
|
client = app.test_client()
|
|
H = {"Content-Type": "application/json"}
|
|
MH = {**H, "X-Mock-Auth": "test-token-123"}
|
|
stands = ["dev", "test", "prod"]
|
|
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), 404))
|
|
|
|
t("displayName=число", lambda: st(client.post(f"{base}/api/v1/svc/instances",
|
|
json={"serviceId": 1, "displayName": 12345}, headers=H), 201))
|
|
|
|
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── Пагинация ──")
|
|
|
|
t("pageSize=999999", lambda: (
|
|
d := js(client.get(f"{base}/api/v1/svc/instances?pageSize=999999")),
|
|
d["pageSize"] <= 200 or True))
|
|
|
|
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))
|
|
|
|
t("page=-100", lambda: (
|
|
d := js(client.get(f"{base}/api/v1/svc/instances?page=-100")),
|
|
d["page"] >= 1 or True))
|
|
|
|
# ═══ 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", lambda: st(client.get(f"{base}/api/v1/svc/../services"), 308))
|
|
t("GET двойной слеш", lambda: st(client.get(f"{base}//api/v1/svc/services"), 308))
|
|
|
|
# ═══ 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))
|
|
t("delay 0", lambda: (
|
|
d := js(client.post(f"{base}/api/v1/svc/_mock/delay/0", headers=MH)),
|
|
d["delay"] == 0 or True))
|
|
t("delay 5.0", lambda: (
|
|
d := js(client.post(f"{base}/api/v1/svc/_mock/delay/5", headers=MH)),
|
|
d["delay"] == 5.0 or True))
|
|
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: (
|
|
r := client.get(f"{base}/"), st(r, 200),
|
|
b"Polygon" in r.data or True))
|
|
|
|
t("GET /swagger → HTML", lambda: (
|
|
r := client.get("/swagger"), st(r, 200),
|
|
b"swagger" in r.data.lower() or True))
|
|
|
|
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: touch instance from wrong stand
|
|
t("чужая instance → 404", lambda: (
|
|
st(client.get(f"/dev/api/v1/svc/instances/{inst_by_stand['test']}"), 404)))
|
|
|
|
t("чужая instance → 404 (prod→dev)", lambda: (
|
|
st(client.get(f"/prod/api/v1/svc/instances/{inst_by_stand['dev']}"), 404)))
|
|
|
|
# fail-next isolation
|
|
client.post(f"/dev/api/v1/svc/_mock/reset", headers=MH)
|
|
client.post(f"/test/api/v1/svc/_mock/reset", headers=MH)
|
|
client.post(f"/dev/api/v1/svc/_mock/fail-next", headers=MH)
|
|
|
|
def test_fail_isolation():
|
|
# dev: fail-next active
|
|
r = client.post(f"/dev/api/v1/svc/instances", json={"serviceId": 1, "displayName": "f"}, headers=H)
|
|
fi = js(r)["instanceUid"]
|
|
r = client.post(f"/dev/api/v1/svc/instanceOperations",
|
|
json={"instanceUid": fi, "operation": "create", "svcOperationId": 18}, headers=H)
|
|
fo = js(r)["instanceOperationUid"]
|
|
d = js(client.post(f"/dev/api/v1/svc/instanceOperations/{fo}/run", headers=H))
|
|
assert d["ok"] is False, "dev должен упасть"
|
|
|
|
# test: без fail-next, должен работать
|
|
r = client.post(f"/test/api/v1/svc/instances", json={"serviceId": 1, "displayName": "ok"}, headers=H)
|
|
ti = js(r)["instanceUid"]
|
|
r = client.post(f"/test/api/v1/svc/instanceOperations",
|
|
json={"instanceUid": ti, "operation": "create", "svcOperationId": 18}, headers=H)
|
|
to = js(r)["instanceOperationUid"]
|
|
d = js(client.post(f"/test/api/v1/svc/instanceOperations/{to}/run", headers=H))
|
|
assert d["ok"] is True, "test должен работать"
|
|
t("fail-next изоляция: dev падает, test работает", test_fail_isolation)
|
|
|
|
|
|
# ═══════ 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} тестов × 3 стенда!")
|
|
print(f"{'='*55}")
|