125 lines
5.2 KiB
Python
125 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
||
"""Smoke-тест моста: stub requests/psycopg2, реальный HTTPServer, проверка
|
||
аутентификации, UUID-валидации и BSL-экранирования."""
|
||
import os, sys, types, json, threading, urllib.request, urllib.error
|
||
|
||
# --- stubs ---
|
||
requests_stub = types.ModuleType("requests")
|
||
captured = {}
|
||
|
||
def fake_post(url, json=None, headers=None, timeout=None):
|
||
captured["code"] = json.get("code", "")
|
||
captured["headers"] = headers
|
||
class R:
|
||
status_code = 200
|
||
text = ""
|
||
def json(self):
|
||
return {"success": True, "data": "[]"}
|
||
return R()
|
||
requests_stub.post = fake_post
|
||
requests_stub.get = lambda *a, **k: None
|
||
|
||
psycopg2_stub = types.ModuleType("psycopg2")
|
||
psycopg2_stub.connect = lambda *a, **k: (_ for _ in ()).throw(Exception("no db in test"))
|
||
|
||
sys.modules["requests"] = requests_stub
|
||
sys.modules["psycopg2"] = psycopg2_stub
|
||
|
||
os.environ["ONEC_MCP_TOKEN"] = "test-mcp-token"
|
||
os.environ["BRIDGE_API_KEY"] = "test-bridge-key"
|
||
os.environ["POSTGRES_PASSWORD"] = "x"
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
import bridge_1c
|
||
|
||
from http.server import HTTPServer
|
||
srv = HTTPServer(("127.0.0.1", 0), bridge_1c.BridgeHandler)
|
||
port = srv.server_address[1]
|
||
t = threading.Thread(target=srv.serve_forever, daemon=True)
|
||
t.start()
|
||
|
||
BASE = f"http://127.0.0.1:{port}"
|
||
KEY = {"X-Api-Key": "test-bridge-key"}
|
||
fails = []
|
||
|
||
def req(method, path, body=None, headers=None):
|
||
from urllib.parse import quote
|
||
data = json.dumps(body).encode() if body is not None else None
|
||
r = urllib.request.Request(BASE + quote(path, safe="/?=&%"), data=data, method=method,
|
||
headers=headers or {}, )
|
||
try:
|
||
with urllib.request.urlopen(r, timeout=5) as resp:
|
||
return resp.status, resp.read()
|
||
except urllib.error.HTTPError as e:
|
||
return e.code, e.read()
|
||
|
||
def check(name, cond, extra=""):
|
||
print(("PASS" if cond else "FAIL"), name, extra)
|
||
if not cond:
|
||
fails.append(name)
|
||
|
||
# 1. без ключа -> 401
|
||
st, _ = req("GET", "/doctors/x/services")
|
||
check("no-auth GET -> 401", st == 401, st)
|
||
st, _ = req("POST", "/appointments", {})
|
||
check("no-auth POST -> 401", st == 401, st)
|
||
|
||
# 2. неверный ключ -> 401
|
||
st, _ = req("GET", "/branches", headers={"X-Api-Key": "wrong"})
|
||
check("bad key -> 401", st == 401, st)
|
||
|
||
# 3. Bearer тоже принимается
|
||
st, _ = req("GET", "/branches", headers={"Authorization": "Bearer test-bridge-key"})
|
||
check("Bearer ok", st == 200, st)
|
||
assert "Bearer" not in json.dumps(captured.get("headers", {})) or True
|
||
check("1C got MCP token", captured["headers"].get("Authorization") == "Bearer test-mcp-token")
|
||
|
||
# 4. UUID-валидация: инъекция в doctor_id
|
||
st, _ = req("GET", '/doctors/abc");Выполнить(0);//services', headers=KEY)
|
||
check("bad doctor_id -> 400", st == 400, st)
|
||
st, _ = req("GET", '/schedule?doctor_id=x%22%3BВыполнить(0)%3B', headers=KEY)
|
||
check("bad schedule doctor_id -> 400", st == 400, st)
|
||
st, _ = req("GET", '/schedule?branch_id=x%22%3BВыполнить(0)%3B', headers=KEY)
|
||
check("bad schedule branch_id -> 400", st == 400, st)
|
||
st, _ = req("GET", '/appointments/not-a-uuid', headers=KEY)
|
||
check("bad appt_id -> 400", st == 400, st)
|
||
|
||
# 5. легитимный uuid проходит к 1С
|
||
st, _ = req("GET", "/doctors/a748a2a7-8771-11ed-840b-00155d46cd22/services", headers=KEY)
|
||
check("good doctor_id -> 200", st == 200, st)
|
||
check("uuid intact in BSL", 'УникальныйИдентификатор("a748a2a7-8771-11ed-840b-00155d46cd22")' in captured["code"])
|
||
|
||
# 6. POST /appointments: инъекция в имени/комментарии экранируется
|
||
evil = 'Ваня"; Результат = "PWNED"; x := "'
|
||
st, body = req("POST", "/appointments", {
|
||
"doctor_id": "a748a2a7-8771-11ed-840b-00155d46cd22",
|
||
"branch_id": "f07031a2-21a6-11f0-8712-00155d46cd22",
|
||
"patient_name": evil,
|
||
"patient_phone": "+7 (999) 123-45-67\" INJECT",
|
||
"comment": 'comment"; Док.Записать(); "',
|
||
"start": "2026-10-20T10:00:00+03:00",
|
||
}, KEY)
|
||
check("appointments accepted", st == 201, st)
|
||
c = captured["code"]
|
||
check("name escaped", 'Наименование = "Ваня""; Результат = ""PWNED""; x := """;' in c)
|
||
check("comment escaped", '"Онлайн-запись: comment""; Док.Записать(); """;' in c)
|
||
check("phone cleaned", 'Представление = "+7 (999) 123-45-67 "' in c)
|
||
# в сгенерированном коде не должно остаться разрыва: ищем подозрительные куски вне строк
|
||
check("no raw PWNED", 'Результат = "PWNED"' not in c.replace('""PWNED""', ""))
|
||
|
||
# 7. невалидные поля -> 400
|
||
st, _ = req("POST", "/appointments", {"doctor_id": 'x";', "start": "2026-10-20T10:00:00"}, KEY)
|
||
check("bad doctor_id POST -> 400", st == 400, st)
|
||
st, _ = req("POST", "/appointments", {"doctor_id": "a748a2a7-8771-11ed-840b-00155d46cd22",
|
||
"start": 'x";Выполнить(0);"'}, KEY)
|
||
check("bad start -> 400", st == 400, st)
|
||
|
||
# 8. healthz без ключа
|
||
st, _ = req("GET", "/healthz")
|
||
check("healthz open", st == 200, st)
|
||
|
||
srv.shutdown()
|
||
print()
|
||
print("FAILURES:", fails if fails else "none")
|
||
sys.exit(1 if fails else 0)
|