@samitouri / QOSami_Agents_Zero-2
Code Commits Issues Pulls Sessions CI/CD
main
main
py 193 lines 5.66 KB
Raw
1 from __future__ import annotations
2
3 from flask import Flask, Response
4
5 import pytest
6
7 from helpers import runtime
8
9
10 def _make_app() -> Flask:
11 app = Flask("test_http_auth_csrf")
12 app.secret_key = "test-secret"
13
14 @app.get("/login")
15 def login_handler():
16 return Response("login", status=200)
17
18 return app
19
20
21 def _set_session(client, **values) -> None:
22 with client.session_transaction() as sess:
23 for key, value in values.items():
24 sess[key] = value
25
26
27 def _set_csrf_cookie(client, token: str) -> None:
28 cookie_name = f"csrf_token_{runtime.get_runtime_id()}"
29 client.set_cookie(cookie_name, token)
30
31
32 def test_http_auth_enforced_when_configured(monkeypatch) -> None:
33 from run_ui import csrf_protect, requires_auth
34
35 monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")
36
37 app = _make_app()
38
39 @app.get("/secure")
40 @requires_auth
41 @csrf_protect
42 async def secure():
43 return Response("ok", status=200)
44
45 client = app.test_client()
46 response = client.get("/secure")
47 assert response.status_code == 302
48
49
50 def test_http_csrf_required_even_when_auth_not_configured(monkeypatch) -> None:
51 from run_ui import csrf_protect, requires_auth
52
53 monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: None)
54
55 app = _make_app()
56
57 @app.get("/secure")
58 @requires_auth
59 @csrf_protect
60 async def secure():
61 return Response("ok", status=200)
62
63 client = app.test_client()
64 _set_session(client, csrf_token="csrf-1")
65 response = client.get("/secure")
66 assert response.status_code == 403
67
68
69 def test_http_csrf_rejects_missing_token(monkeypatch) -> None:
70 from run_ui import csrf_protect, requires_auth
71
72 monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")
73
74 app = _make_app()
75
76 @app.get("/secure")
77 @requires_auth
78 @csrf_protect
79 async def secure():
80 return Response("ok", status=200)
81
82 client = app.test_client()
83 _set_session(client, authentication="hash", csrf_token="csrf-2")
84 response = client.get("/secure")
85 assert response.status_code == 403
86
87
88 def test_http_csrf_accepts_valid_header_without_cookie(monkeypatch) -> None:
89 from run_ui import csrf_protect, requires_auth
90
91 monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")
92
93 app = _make_app()
94
95 @app.get("/secure")
96 @requires_auth
97 @csrf_protect
98 async def secure():
99 return Response("ok", status=200)
100
101 client = app.test_client()
102 _set_session(client, authentication="hash", csrf_token="csrf-3")
103 response = client.get("/secure", headers={"X-CSRF-Token": "csrf-3"})
104 assert response.status_code == 200
105
106
107 def test_http_csrf_accepts_valid_cookie(monkeypatch) -> None:
108 from run_ui import csrf_protect, requires_auth
109
110 monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")
111
112 app = _make_app()
113
114 @app.get("/secure")
115 @requires_auth
116 @csrf_protect
117 async def secure():
118 return Response("ok", status=200)
119
120 client = app.test_client()
121 _set_session(client, authentication="hash", csrf_token="csrf-4")
122 _set_csrf_cookie(client, "csrf-4")
123 response = client.get("/secure")
124 assert response.status_code == 200
125
126
127 def test_safe_next_url_accepts_plugin_page_path() -> None:
128 from helpers.api import get_safe_next_url, is_safe_next_url
129
130 target = "/plugins/a0_voqualizer/webui/voqualizer.html"
131 assert is_safe_next_url(target)
132 assert get_safe_next_url(target, "/") == target
133
134
135 def test_safe_next_url_preserves_query_string() -> None:
136 from helpers.api import get_safe_next_url
137
138 target = "/plugins/a0_voqualizer/webui/voqualizer.html?context=rlO1iMV7"
139 assert get_safe_next_url(target, "/") == target
140
141
142 def test_safe_next_url_rejects_external_and_protocol_relative_urls() -> None:
143 from helpers.api import get_safe_next_url, is_safe_next_url
144
145 fallback = "/"
146 for value in [
147 "https://evil.example/plugins/a0_voqualizer/webui/voqualizer.html",
148 "//evil.example/plugins/a0_voqualizer/webui/voqualizer.html",
149 "javascript:alert(1)",
150 "/safe\nLocation: https://evil.example",
151 ]:
152 assert not is_safe_next_url(value)
153 assert get_safe_next_url(value, fallback) == fallback
154
155
156 def test_auth_redirect_includes_original_path_and_query(monkeypatch) -> None:
157 from run_ui import requires_auth
158
159 monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")
160
161 app = _make_app()
162
163 @app.get("/plugins/a0_voqualizer/webui/voqualizer.html")
164 @requires_auth
165 async def voqualizer_page():
166 return Response("ok", status=200)
167
168 client = app.test_client()
169 response = client.get("/plugins/a0_voqualizer/webui/voqualizer.html?context=rlO1iMV7")
170 assert response.status_code == 302
171 location = response.headers["Location"]
172 assert location.startswith("/login?next=")
173 assert "%2Fplugins%2Fa0_voqualizer%2Fwebui%2Fvoqualizer.html%3Fcontext%3DrlO1iMV7" in location
174
175
176 def test_is_safe_next_url_rejects_backslash_open_redirects() -> None:
177 from helpers.api import is_safe_next_url
178
179 # Raw backslash forms
180 assert is_safe_next_url("/\\evil.example") is False
181 assert is_safe_next_url("\\/evil.example") is False
182 assert is_safe_next_url("/path\\evil") is False
183
184 # Percent-encoded backslash forms
185 assert is_safe_next_url("/%5Cevil.example") is False
186 assert is_safe_next_url("%5C/evil.example") is False
187 assert is_safe_next_url("/%5cevil.example") is False # lowercase hex
188
189 # Mixed / double-encoded edge
190 assert is_safe_next_url("/path/%5Cevil") is False
191
192 # Sanity: a legitimate relative path still passes
193 assert is_safe_next_url("/plugins/a0_voqualizer/webui/voqualizer.html") is True