247 lines
8.4 KiB
Python
247 lines
8.4 KiB
Python
import base64
|
|
import json
|
|
|
|
import pytest
|
|
import jwt
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
from app.api.meta import router as meta_router
|
|
from app.api import captures
|
|
from app.api.deps import create_session, session_matches, session_secret
|
|
from app.api.routes import router as api_router
|
|
from app.api.schwab_auth import router as schwab_auth_router
|
|
from app.api.ws import router as ws_router
|
|
from app.config import Settings
|
|
from app.runtime import Runtime
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path):
|
|
def build(token: str, password: str = "") -> TestClient:
|
|
settings = Settings(
|
|
chart_auth_token=token,
|
|
chart_password=password,
|
|
manual_lines_path=tmp_path / "manual_lines.json",
|
|
)
|
|
app = FastAPI()
|
|
app.include_router(meta_router)
|
|
app.include_router(schwab_auth_router)
|
|
app.include_router(api_router)
|
|
app.include_router(ws_router)
|
|
app.state.runtime = Runtime(settings)
|
|
return TestClient(app)
|
|
|
|
return build
|
|
|
|
|
|
def test_open_when_no_token_configured(client):
|
|
assert client("").get("/api/bars").status_code == 200
|
|
|
|
|
|
def test_oauth_callback_stays_open_when_a_token_is_set():
|
|
# The provider redirects a browser here and cannot attach the chart token.
|
|
# A 401 would break the login flow at its last step, on production only,
|
|
# where CHART_AUTH_TOKEN is the one thing that differs from local.
|
|
from fastapi import FastAPI as _FastAPI
|
|
from app.config import Settings as _Settings
|
|
from app.runtime import Runtime as _Runtime
|
|
import tempfile, pathlib as _pathlib
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
app = _FastAPI()
|
|
app.include_router(meta_router)
|
|
app.include_router(schwab_auth_router)
|
|
app.include_router(api_router)
|
|
app.state.runtime = _Runtime(
|
|
_Settings(chart_auth_token="s3cret",
|
|
manual_lines_path=_pathlib.Path(tmp) / "manual_lines.json")
|
|
)
|
|
probe = TestClient(app)
|
|
assert probe.get("/api/qt").status_code == 200
|
|
assert probe.get("/api/status").status_code == 401
|
|
|
|
|
|
def test_rejects_missing_token(client):
|
|
assert client("s3cret").get("/api/bars").status_code == 401
|
|
|
|
|
|
def test_rejects_wrong_token(client):
|
|
response = client("s3cret").get("/api/bars", headers={"X-Chart-Token": "nope"})
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_accepts_header_token(client):
|
|
response = client("s3cret").get("/api/bars", headers={"X-Chart-Token": "s3cret"})
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_accepts_query_token(client):
|
|
assert client("s3cret").get("/api/bars?token=s3cret").status_code == 200
|
|
|
|
|
|
def test_password_login_uses_an_httponly_session_instead_of_exposing_the_token(client):
|
|
probe = client("opaque-api-token", "friendly passphrase")
|
|
|
|
response = probe.post("/api/login", json={"password": "friendly passphrase"})
|
|
|
|
assert response.status_code == 204
|
|
cookie = response.headers["set-cookie"]
|
|
assert "chart-session=" in cookie
|
|
assert "HttpOnly" in cookie
|
|
assert "SameSite=strict" in cookie
|
|
assert "opaque-api-token" not in cookie
|
|
assert probe.get("/api/bars").status_code == 200
|
|
|
|
|
|
def test_wrong_password_cannot_create_a_session(client):
|
|
probe = client("opaque-api-token", "friendly passphrase")
|
|
|
|
assert probe.post("/api/login", json={"password": "wrong"}).status_code == 401
|
|
assert probe.get("/api/bars").status_code == 401
|
|
|
|
|
|
def test_existing_browser_token_is_exchanged_for_a_session(client):
|
|
probe = client("opaque-api-token", "friendly passphrase")
|
|
|
|
response = probe.post(
|
|
"/api/login",
|
|
json={"password": ""},
|
|
headers={"X-Chart-Token": "opaque-api-token"},
|
|
)
|
|
|
|
assert response.status_code == 204
|
|
assert probe.get("/api/bars").status_code == 200
|
|
|
|
|
|
def test_logout_invalidates_the_browser_session(client):
|
|
probe = client("opaque-api-token", "friendly passphrase")
|
|
assert probe.post("/api/login", json={"password": "friendly passphrase"}).status_code == 204
|
|
|
|
assert probe.post("/api/logout").status_code == 204
|
|
assert probe.get("/api/bars").status_code == 401
|
|
|
|
|
|
def test_session_signature_and_expiry_are_enforced(client):
|
|
probe = client("opaque-api-token", "friendly passphrase")
|
|
session = create_session(probe.app)
|
|
expired = jwt.encode(
|
|
{"sub": "shared", "iat": 1, "exp": 2},
|
|
session_secret(probe.app),
|
|
algorithm="HS256",
|
|
)
|
|
|
|
assert session_matches(probe.app, session)
|
|
assert not session_matches(probe.app, f"{session}tampered")
|
|
assert not session_matches(probe.app, expired)
|
|
|
|
|
|
def test_unicode_passwords_do_not_crash_login(client):
|
|
probe = client("opaque-api-token", "correct horse ünicode")
|
|
|
|
assert probe.post("/api/login", json={"password": "wrong pässword"}).status_code == 401
|
|
assert probe.post("/api/login", json={"password": "correct horse ünicode"}).status_code == 204
|
|
|
|
|
|
def test_https_login_marks_the_session_cookie_secure(client):
|
|
response = client("opaque-api-token", "friendly passphrase").post(
|
|
"/api/login",
|
|
json={"password": "friendly passphrase"},
|
|
headers={"X-Forwarded-Proto": "https"},
|
|
)
|
|
|
|
assert "Secure" in response.headers["set-cookie"]
|
|
|
|
|
|
def test_password_alone_enables_auth(client):
|
|
probe = client("", "friendly passphrase")
|
|
|
|
assert probe.get("/api/bars").status_code == 401
|
|
assert probe.post("/api/login", json={"password": "friendly passphrase"}).status_code == 204
|
|
assert probe.get("/api/bars").status_code == 200
|
|
|
|
|
|
def test_writes_are_protected(client):
|
|
payload = {
|
|
"tf": "1m",
|
|
"side": "support",
|
|
"anchor_t": 1,
|
|
"anchor_p": 1.0,
|
|
"end_t": 2,
|
|
"end_p": 2.0,
|
|
}
|
|
assert client("s3cret").post("/api/lines", json=payload).status_code == 401
|
|
|
|
|
|
def test_diagnostic_capture_is_authenticated_and_retrievable_by_capability_id(
|
|
client, tmp_path, monkeypatch
|
|
):
|
|
monkeypatch.setattr(captures, "CAPTURE_DIR", tmp_path / "captures")
|
|
probe = client("s3cret")
|
|
image = b"\x89PNG\r\n\x1a\ntrimmed-test-image"
|
|
metadata = base64.b64encode(
|
|
json.dumps({"timeframe": "30m", "viewport_width": 1440}).encode()
|
|
).decode()
|
|
|
|
assert probe.post(
|
|
"/api/debug/captures",
|
|
content=image,
|
|
headers={"Content-Type": "image/png", "X-Capture-Metadata": metadata},
|
|
).status_code == 401
|
|
response = probe.post(
|
|
"/api/debug/captures",
|
|
content=image,
|
|
headers={
|
|
"X-Chart-Token": "s3cret",
|
|
"Content-Type": "image/png",
|
|
"X-Capture-Metadata": metadata,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
saved = response.json()
|
|
assert saved["id"].startswith("c-")
|
|
assert probe.get(saved["url"]).content == image
|
|
details = probe.get(saved["metadata_url"]).json()
|
|
assert details["id"] == saved["id"]
|
|
assert details["timeframe"] == "30m"
|
|
assert details["viewport_width"] == 1440
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["/api/health", "/api/version"])
|
|
def test_meta_endpoints_stay_open(client, path):
|
|
"""bin/wait-deploy polls /api/version without carrying the token."""
|
|
assert client("s3cret").get(path).status_code == 200
|
|
|
|
|
|
def test_websocket_rejects_missing_token(client):
|
|
with pytest.raises(WebSocketDisconnect) as excinfo:
|
|
with client("s3cret").websocket_connect("/ws"):
|
|
pass
|
|
assert excinfo.value.code == 1008
|
|
|
|
|
|
def test_websocket_accepts_query_token(client):
|
|
with client("s3cret").websocket_connect("/ws?token=s3cret") as socket:
|
|
assert socket.receive_json()["type"] == "snapshot"
|
|
|
|
|
|
def test_websocket_accepts_the_password_session_cookie(client):
|
|
probe = client("opaque-api-token", "friendly passphrase")
|
|
assert probe.post("/api/login", json={"password": "friendly passphrase"}).status_code == 204
|
|
|
|
with probe.websocket_connect("/ws", headers={"origin": "http://testserver"}) as socket:
|
|
assert socket.receive_json()["type"] == "snapshot"
|
|
|
|
|
|
def test_websocket_rejects_a_session_cookie_from_another_origin(client):
|
|
probe = client("opaque-api-token", "friendly passphrase")
|
|
assert probe.post("/api/login", json={"password": "friendly passphrase"}).status_code == 204
|
|
|
|
with pytest.raises(WebSocketDisconnect) as excinfo:
|
|
with probe.websocket_connect(
|
|
"/ws", headers={"origin": "https://other.example.com"}
|
|
):
|
|
pass
|
|
assert excinfo.value.code == 1008
|