File size: 5,793 Bytes
be6c5ee
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
from __future__ import annotations

from flask import Flask, Response

import pytest

from helpers import runtime


def _make_app() -> Flask:
    app = Flask("test_http_auth_csrf")
    app.secret_key = "test-secret"

    @app.get("/login")
    def login_handler():
        return Response("login", status=200)

    return app


def _set_session(client, **values) -> None:
    with client.session_transaction() as sess:
        for key, value in values.items():
            sess[key] = value


def _set_csrf_cookie(client, token: str) -> None:
    cookie_name = f"csrf_token_{runtime.get_runtime_id()}"
    client.set_cookie(cookie_name, token)


def test_http_auth_enforced_when_configured(monkeypatch) -> None:
    from run_ui import csrf_protect, requires_auth

    monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")

    app = _make_app()

    @app.get("/secure")
    @requires_auth
    @csrf_protect
    async def secure():
        return Response("ok", status=200)

    client = app.test_client()
    response = client.get("/secure")
    assert response.status_code == 302


def test_http_csrf_required_even_when_auth_not_configured(monkeypatch) -> None:
    from run_ui import csrf_protect, requires_auth

    monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: None)

    app = _make_app()

    @app.get("/secure")
    @requires_auth
    @csrf_protect
    async def secure():
        return Response("ok", status=200)

    client = app.test_client()
    _set_session(client, csrf_token="csrf-1")
    response = client.get("/secure")
    assert response.status_code == 403


def test_http_csrf_rejects_missing_token(monkeypatch) -> None:
    from run_ui import csrf_protect, requires_auth

    monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")

    app = _make_app()

    @app.get("/secure")
    @requires_auth
    @csrf_protect
    async def secure():
        return Response("ok", status=200)

    client = app.test_client()
    _set_session(client, authentication="hash", csrf_token="csrf-2")
    response = client.get("/secure")
    assert response.status_code == 403


def test_http_csrf_accepts_valid_header_without_cookie(monkeypatch) -> None:
    from run_ui import csrf_protect, requires_auth

    monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")

    app = _make_app()

    @app.get("/secure")
    @requires_auth
    @csrf_protect
    async def secure():
        return Response("ok", status=200)

    client = app.test_client()
    _set_session(client, authentication="hash", csrf_token="csrf-3")
    response = client.get("/secure", headers={"X-CSRF-Token": "csrf-3"})
    assert response.status_code == 200


def test_http_csrf_accepts_valid_cookie(monkeypatch) -> None:
    from run_ui import csrf_protect, requires_auth

    monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")

    app = _make_app()

    @app.get("/secure")
    @requires_auth
    @csrf_protect
    async def secure():
        return Response("ok", status=200)

    client = app.test_client()
    _set_session(client, authentication="hash", csrf_token="csrf-4")
    _set_csrf_cookie(client, "csrf-4")
    response = client.get("/secure")
    assert response.status_code == 200


def test_safe_next_url_accepts_plugin_page_path() -> None:
    from helpers.api import get_safe_next_url, is_safe_next_url

    target = "/plugins/a0_voqualizer/webui/voqualizer.html"
    assert is_safe_next_url(target)
    assert get_safe_next_url(target, "/") == target


def test_safe_next_url_preserves_query_string() -> None:
    from helpers.api import get_safe_next_url

    target = "/plugins/a0_voqualizer/webui/voqualizer.html?context=rlO1iMV7"
    assert get_safe_next_url(target, "/") == target


def test_safe_next_url_rejects_external_and_protocol_relative_urls() -> None:
    from helpers.api import get_safe_next_url, is_safe_next_url

    fallback = "/"
    for value in [
        "https://evil.example/plugins/a0_voqualizer/webui/voqualizer.html",
        "//evil.example/plugins/a0_voqualizer/webui/voqualizer.html",
        "javascript:alert(1)",
        "/safe\nLocation: https://evil.example",
    ]:
        assert not is_safe_next_url(value)
        assert get_safe_next_url(value, fallback) == fallback


def test_auth_redirect_includes_original_path_and_query(monkeypatch) -> None:
    from run_ui import requires_auth

    monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")

    app = _make_app()

    @app.get("/plugins/a0_voqualizer/webui/voqualizer.html")
    @requires_auth
    async def voqualizer_page():
        return Response("ok", status=200)

    client = app.test_client()
    response = client.get("/plugins/a0_voqualizer/webui/voqualizer.html?context=rlO1iMV7")
    assert response.status_code == 302
    location = response.headers["Location"]
    assert location.startswith("/login?next=")
    assert "%2Fplugins%2Fa0_voqualizer%2Fwebui%2Fvoqualizer.html%3Fcontext%3DrlO1iMV7" in location


def test_is_safe_next_url_rejects_backslash_open_redirects() -> None:
    from helpers.api import is_safe_next_url

    # Raw backslash forms
    assert is_safe_next_url("/\\evil.example") is False
    assert is_safe_next_url("\\/evil.example") is False
    assert is_safe_next_url("/path\\evil") is False

    # Percent-encoded backslash forms
    assert is_safe_next_url("/%5Cevil.example") is False
    assert is_safe_next_url("%5C/evil.example") is False
    assert is_safe_next_url("/%5cevil.example") is False  # lowercase hex

    # Mixed / double-encoded edge
    assert is_safe_next_url("/path/%5Cevil") is False

    # Sanity: a legitimate relative path still passes
    assert is_safe_next_url("/plugins/a0_voqualizer/webui/voqualizer.html") is True