155 lines
5.6 KiB
Python
155 lines
5.6 KiB
Python
from __future__ import annotations
|
|
|
|
from flask import Flask, jsonify
|
|
|
|
from atlas_portal.routes import account_wolf
|
|
|
|
|
|
class DummyAriadne:
|
|
def __init__(self, enabled: bool = True) -> None:
|
|
self._enabled = enabled
|
|
self.calls: list[tuple[str, str, object | None, dict | None]] = []
|
|
|
|
def enabled(self) -> bool:
|
|
return self._enabled
|
|
|
|
def proxy(self, method: str, path: str, payload: object | None = None, params: dict | None = None):
|
|
self.calls.append((method, path, payload, params))
|
|
return jsonify({"path": path, "payload": payload, "params": params})
|
|
|
|
|
|
def make_client(monkeypatch, *, ariadne: DummyAriadne | None = None, account_ok: bool = True):
|
|
app = Flask(__name__)
|
|
active_ariadne = ariadne or DummyAriadne()
|
|
monkeypatch.setattr(account_wolf, "ariadne_client", active_ariadne)
|
|
monkeypatch.setattr(account_wolf, "require_auth", lambda fn: fn)
|
|
monkeypatch.setattr(
|
|
account_wolf,
|
|
"require_account_access",
|
|
lambda: (True, None) if account_ok else (False, (jsonify({"error": "forbidden"}), 403)),
|
|
)
|
|
account_wolf.register_account_wolf(app)
|
|
return app.test_client(), active_ariadne
|
|
|
|
|
|
def test_wolf_status_proxies_source_ip(monkeypatch) -> None:
|
|
client, ariadne = make_client(monkeypatch)
|
|
|
|
resp = client.get(
|
|
"/api/account/wolf",
|
|
environ_base={"REMOTE_ADDR": "1.2.3.4"},
|
|
headers={"X-Forwarded-For": "9.9.9.9, 181.1.87.186, 10.0.0.1"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert ariadne.calls == [("GET", "/api/game-stream/status", None, {"source_ip": "1.2.3.4"})]
|
|
|
|
|
|
def test_wolf_status_prefers_public_query_ip(monkeypatch) -> None:
|
|
client, ariadne = make_client(monkeypatch)
|
|
|
|
resp = client.get(
|
|
"/api/account/wolf?source_ip=181.1.87.186",
|
|
environ_base={"REMOTE_ADDR": "10.42.28.0"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert ariadne.calls == [("GET", "/api/game-stream/status", None, {"source_ip": "181.1.87.186"})]
|
|
|
|
|
|
def test_wolf_unlock_uses_current_source_ip(monkeypatch) -> None:
|
|
client, ariadne = make_client(monkeypatch)
|
|
|
|
resp = client.post(
|
|
"/api/account/wolf/firewall/unlock",
|
|
json={"ttl_seconds": 120},
|
|
environ_base={"REMOTE_ADDR": "5.6.7.8"},
|
|
headers={"X-Real-IP": "10.42.28.0"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert ariadne.calls == [("POST", "/api/game-stream/firewall/unlock", {"ttl_seconds": 120, "ip": "5.6.7.8"}, None)]
|
|
|
|
|
|
def test_wolf_unlock_prefers_public_payload_ip(monkeypatch) -> None:
|
|
client, ariadne = make_client(monkeypatch)
|
|
|
|
resp = client.post(
|
|
"/api/account/wolf/firewall/unlock",
|
|
json={"ip": "181.1.87.186"},
|
|
environ_base={"REMOTE_ADDR": "10.42.28.0"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert ariadne.calls == [("POST", "/api/game-stream/firewall/unlock", {"ip": "181.1.87.186"}, None)]
|
|
|
|
|
|
def test_wolf_unlock_ignores_private_payload_ip(monkeypatch) -> None:
|
|
client, ariadne = make_client(monkeypatch)
|
|
|
|
resp = client.post(
|
|
"/api/account/wolf/firewall/unlock",
|
|
json={"ip": "10.0.0.5"},
|
|
environ_base={"REMOTE_ADDR": "5.6.7.8"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert ariadne.calls == [("POST", "/api/game-stream/firewall/unlock", {"ip": "5.6.7.8"}, None)]
|
|
|
|
|
|
def test_wolf_source_ip_prefers_nearest_public_proxy_value(monkeypatch) -> None:
|
|
client, ariadne = make_client(monkeypatch)
|
|
|
|
resp = client.post(
|
|
"/api/account/wolf/firewall/unlock",
|
|
json={},
|
|
environ_base={"REMOTE_ADDR": "10.42.28.0"},
|
|
headers={"X-Forwarded-For": "9.9.9.9, 181.1.87.186, 10.42.28.0"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert ariadne.calls == [("POST", "/api/game-stream/firewall/unlock", {"ip": "181.1.87.186"}, None)]
|
|
|
|
|
|
def test_wolf_pairing_and_admin_actions_proxy(monkeypatch) -> None:
|
|
client, ariadne = make_client(monkeypatch)
|
|
|
|
client.get("/api/account/wolf/pairing/status?source_ip=181.1.87.186", environ_base={"REMOTE_ADDR": "1.2.3.4"})
|
|
client.post(
|
|
"/api/account/wolf/pairing/submit-pin",
|
|
json={"pair_secret": "secret", "pin": "1234", "source_ip": "181.1.87.186"},
|
|
environ_base={"REMOTE_ADDR": "1.2.3.4"},
|
|
)
|
|
client.post("/api/account/wolf/game-mode/start", json={"game": "steam"})
|
|
client.post("/api/account/wolf/game-mode/stop", json={"game": "steam"})
|
|
client.post("/api/account/wolf/admin/firewall/unlock", json={"ip": "8.8.8.8", "target_user": "olya"})
|
|
|
|
assert ariadne.calls == [
|
|
("GET", "/api/game-stream/pairing/status", None, {"source_ip": "181.1.87.186"}),
|
|
("POST", "/api/game-stream/pairing/submit-pin", {"pair_secret": "secret", "pin": "1234", "source_ip": "181.1.87.186"}, None),
|
|
("POST", "/api/admin/game-mode/start", {"game": "steam"}, None),
|
|
("POST", "/api/admin/game-mode/stop", {"game": "steam"}, None),
|
|
("POST", "/api/admin/game-stream/firewall/unlock", {"ip": "8.8.8.8", "target_user": "olya"}, None),
|
|
]
|
|
|
|
|
|
def test_wolf_user_revoke_accepts_public_payload_ip(monkeypatch) -> None:
|
|
client, ariadne = make_client(monkeypatch)
|
|
|
|
resp = client.post(
|
|
"/api/account/wolf/firewall/revoke",
|
|
json={"ip": "9.9.9.9"},
|
|
environ_base={"REMOTE_ADDR": "1.2.3.4"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert ariadne.calls == [("POST", "/api/game-stream/firewall/revoke", {"ip": "9.9.9.9"}, None)]
|
|
|
|
|
|
def test_wolf_routes_require_account_and_ariadne(monkeypatch) -> None:
|
|
client, _ariadne = make_client(monkeypatch, account_ok=False)
|
|
assert client.get("/api/account/wolf").status_code == 403
|
|
|
|
client, _ariadne = make_client(monkeypatch, ariadne=DummyAriadne(enabled=False))
|
|
assert client.get("/api/account/wolf").status_code == 503
|