mirror/sabnzbd
1
0 Fork 0
mirror of https://github.com/sabnzbd/sabnzbd.git synced 2026-09-22 10:55:33 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
feature/scriptlog
sabnzbd/tests/test_interface.py
T

950 lines
48 KiB
Python
Raw Permalink Blame History

#!/usr/bin/python3 -OO
# Copyright 2007-2026 by The SABnzbd-Team (sabnzbd.org)
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
"""
tests.test_interface - Testing functions in interface.py
"""
import inspect
import logging
import logging.config
from typing import Optional
import pytest
from unittest.mock import Mock, patch
from starlette.requests import Request
from starlette.datastructures import Headers, Address, QueryParams
import uvicorn
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
from uvicorn.lifespan import on as lifespan_on
from uvicorn.protocols.http import h11_impl, httptools_impl
from uvicorn.server import ServerState
import sabnzbd
import sabnzbd.cfg as cfg
import sabnzbd.api as api
import sabnzbd.security as security
from sabnzbd import interface
from tests.test_security import (
anonymous_csrf,
api_request,
config_save_middleware,
mock_request,
page_post,
proxied_request,
set_csrf_header,
store_session,
)
from sabnzbd.misc import is_local_addr, is_loopback_addr, xff_trusted_networks
from tests.testhelper import run_async
def resolve_client(remote_ip: str, xff_header: str | None = None, remote_port: int = 12345) -> Address:
"""Pass a connection through ProxyTrustMiddleware, wired exactly as create_app()
does, and return the resulting effective client."""
captured = {}
async def asgi_app(scope, receive, send):
captured["client"] = scope.get("client")
middleware = security.ProxyTrustMiddleware(asgi_app)
headers = []
if xff_header:
headers.append((b"x-forwarded-for", xff_header.encode("latin1")))
scope = {"type": "http", "client": (remote_ip, remote_port), "headers": headers}
run_async(middleware(scope, None, None))
return Address(*captured["client"])
class TestInterfaceFunctions:
@pytest.mark.parametrize(
"remote_ip, local_ranges, xff_header, result_with_xff",
[
("10.11.12.13", None, None, True),
("10.11.12.13", None, "127.0.0.1", True),
("10.11.12.13", None, "127.1.2.3", True),
("10.11.12.13", None, "127.0.0.1:8080", True), # Port stripped from XFF, leaving loopback
("10.11.12.13", None, "::1", True),
("10.11.12.13", None, "[::1]", True),
("10.11.12.13", None, "[::1]:8080", True), # Port stripped from XFF, leaving loopback
("10.11.12.13", None, "localhost", False), # Hostname in XFF
("10.11.12.13", None, "example.org", False), # Hostname in XFF
("10.11.12.13", None, "192.168.1.1", True),
("10.11.12.13", None, "10.11.12.99", True),
("10.11.12.13", None, "8.7.6.5", False), # XFF IP isn't local
("10.11.12.13", None, "192.168.1.1, 10.11.12.13", True),
("10.11.12.13", None, "192.168.1.1, 10.11.12.13, 9.8.7.6", False), # Last XFF IP isn't local
("10.11.12.13", None, "192.168.1.1, 10.11.12.13, ::1", True),
("10.11.12.13", None, "192.168.1.1, 10.11.12.13, sabrules.example.org", False), # Hostname in XFF
("10.11.12.13", "192.168.1.0/24", None, False), # Remote IP not part of local ranges
("10.11.12.13", "192.168.1.0/24", "192.168.1.23", False),
("10.11.12.13", "192.168.1.0/24", "192.168.1.23, 10.11.12.1", False),
("10.11.12.13", "192.168.1.0/24, 10.0.0.0/8", "192.168.1.23", True),
("10.11.12.13", "192.168.2.0/24, 10.0.0.0/8", "192.168.1.23", False),
("10.11.12.13", "192.168.1.0/24, 10.0.0.0/24", "192.168.1.23", False),
("10.11.12.13", "10.11.12.0/24", "192.168.1.23", False),
("10.11.12.13", "2001:ffff::/64", None, False),
("10.11.12.13", "2001:ffff::/64, 192.168.1.0/24", None, False),
("13.12.11.10", None, None, False), # Public remote IP doesn't have access, XFF ignored altogether
("13.12.11.10", None, "127.0.0.1", False),
("13.12.11.10", None, "127.1.2.3", False),
("13.12.11.10", None, "::1", False),
("13.12.11.10", None, "[::1]", False),
("13.12.11.10", None, "localhost", False),
("13.12.11.10", None, "192.168.1.1", False),
("13.12.11.10", None, "192.168.1.1, 13.12.11.10", False),
("13.12.11.10", None, "192.168.1.1, 13.12.11.10, ::1", False),
("13.12.11.10", None, "2001::/16", False),
("13.12.11.10", None, "2001::/16, 13.12.11.10", False),
("13.12.11.10", None, "2001::/16, 13.0.0.0/9", False),
("13.12.11.10", "13.12.11.10", None, True), # Local ranges include a public IP
("13.12.11.10", "13.12.11.10, 192.168.255.0/24", None, True),
("13.12.11.10", "13.12.11.10", "192.168.1.1", False), # XFF not in local ranges
("13.12.11.10", "13.12.11.10, 192.168.255.0/24", "192.168.1.1", False),
("13.12.11.10", "13.12.11.10", "192.168.1.1, 9.8.7.6", False),
("13.12.11.10", "13.12.11.10, 192.168.255.0/24", "192.168.1.1, 9.8.7.6", False),
("13.12.11.10", "13.0.0.0/12", None, True),
("13.12.11.10", "13.0.0.0/12, 192.168.255.0/24", None, True),
("13.12.11.10", "13.0.0.0/12", "192.168.1.1", False), # XFF not in local ranges
("13.12.11.10", "13.0.0.0/12, 192.168.255.0/24", "192.168.1.1", False),
("13.12.11.10", "13.0.0.0/12", "192.168.1.1, 9.8.7.6", False),
("13.12.11.10", "13.0.0.0/12, 192.168.255.0/24", "192.168.1.1, 9.8.7.6", False),
("127.6.6.6", None, None, True),
("127.6.6.6", None, "127.0.0.1", True),
("127.6.6.6", None, "127.1.2.3", True),
("127.6.6.6", None, "127.0.0.1:8080", True), # Port stripped from XFF, leaving loopback
("127.6.6.6", None, "::1", True),
("127.6.6.6", None, "[::1]", True),
("127.6.6.6", None, "[::1]:8080", True), # Port stripped from XFF, leaving loopback
("127.6.6.6", None, "localhost", False), # Hostname in XFF
("127.6.6.6", None, "example.org", False), # Hostname in XFF
("127.6.6.6", None, "192.168.1.1", True),
("127.6.6.6", None, "10.11.12.99", True),
("127.6.6.6", None, "8.7.6.5", False), # XFF IP isn't local
("127.6.6.6", None, "192.168.1.1, 127.6.6.6", True),
("127.6.6.6", None, "192.168.1.1, 127.6.6.6, 9.8.7.6", False), # Last XFF IP isn't local
("127.6.6.6", None, "192.168.1.1, 127.6.6.6, ::1", True),
("127.6.6.6", None, "192.168.1.1, 127.6.6.6, sabrules.example.org", False), # Hostname in XFF
("127.6.6.6", "192.168.1.0/24", None, True), # Remote IP is loopback, local ranges be damned
("127.6.6.6", "192.168.1.0/24", "192.168.1.23", True),
("127.6.6.6", "192.168.1.0/24", "192.168.1.23, 127.0.0.1", True),
("127.6.6.6", "192.168.1.0/24, 127.0.0.0/8", "192.168.1.23", True),
("127.6.6.6", "192.168.2.0/24, 127.0.0.0/8", "192.168.1.23", False), # Access denied by XFF
("127.6.6.6", "192.168.2.0/24, 127.0.0.0/8", "5.6.7.8", False), # Idem
("127.6.6.6", "192.168.1.0/24, 127.0.0.0/8", "192.168.1.23, 5.6.7.8", False), # Idem
("127.6.6.6", "192.168.1.0/24, 10.0.0.0/24", "::1", True),
("127.6.6.6", "127.6.6.0/24", "192.168.1.23", False), # Access denied by XFF
("127.6.6.6", "2001:ffff::/32", None, True),
("127.6.6.6", "2001:ffff::/32, 192.168.1.0/24", None, True),
("127.6.6.6", "2001:ffff::/32", "2001:ffff:a:b:c:d:e:f", True),
("127.6.6.6", "2001:ffff::/32, 192.168.1.0/24", "2001:ffff:a:b:c:d:e:f, 192.168.1.1", True),
("127.6.6.6", "2001:ffff::/32", "666:ffff:a:b:c:d:e:f", False), # Access denied by XFF
("127.6.6.6", "2001:ffff::/32, 192.168.1.0/24", "666:ffff:a:b:c:d:e:f, 192.168.1.1", False), # Idem
("DEAD:BEEF:2023:007::1", None, None, False), # Back to ignoring XFF altogether
("DEAD:BEEF:2023:007::1", None, "127.0.0.1", False), # XFF is loopback
("DEAD:BEEF:2023:007::1", None, "127.1.2.3", False),
("DEAD:BEEF:2023:007::1", None, "::1", False),
("DEAD:BEEF:2023:007::1", None, "[::1]", False),
("DEAD:BEEF:2023:007::1", None, "localhost", False), # Hostname in XFF
("DEAD:BEEF:2023:007::1", None, "192.168.1.1", False),
("DEAD:BEEF:2023:007::1", None, "192.168.1.1, DEAD:BEEF:2023:0007::1", False),
("DEAD:BEEF:2023:007::1", None, "192.168.1.1, DEAD:BEEF:2023:0007::1, ::1", False),
("DEAD:BEEF:2023:007::1", None, "2001::/16", False),
("DEAD:BEEF:2023:007::1", "dead:beef::/32", None, True), # Local ranges include a public IPv6
# ...which is local, but not a local network, so every row below carries a
# header it is not trusted to send
("DEAD:BEEF:2023:007::1", "dead:beef::/32", "127.0.0.1", False),
("DEAD:BEEF:2023:007::1", "dead:beef::/32", "127.1.2.3", False),
("DEAD:BEEF:2023:007::1", "dead:beef::/32", "::1", False),
("DEAD:BEEF:2023:007::1", "dead:beef::/32", "[::1]", False),
("DEAD:BEEF:2023:007::1", "dead:beef::/32", "localhost", False), # Hostname in XFF
("DEAD:BEEF:2023:007::1", "dead:beef::/32", "192.168.1.1", False),
("DEAD:BEEF:2023:007::1", "dead:beef::/32", "192.168.1.1, DEAD:BEEF:2023:0007::1", False),
("DEAD:BEEF:2023:007::1", "dead:beef::/32", "192.168.1.1, DEAD:BEEF:2023:0007::1, ::1", False),
("DEAD:BEEF:2023:007::1", "dead:beef::/32", "DEAD::/16", False), # Netmask in XFF
("DEAD:BEEF:2023:007::1", "dead:beef::/32", "DEAD:BEEF:2023:7::42", False),
],
)
@pytest.mark.parametrize("access_type", [1, 2, 3, 4, 5, 6])
@pytest.mark.parametrize("inet_exposure", [0, 1, 2, 3, 4, 5])
@pytest.mark.parametrize("verify_xff_header", [False, True])
@pytest.mark.config(
lambda params: {
"local_ranges": params["local_ranges"],
"inet_exposure": params["inet_exposure"],
"verify_xff_header": params["verify_xff_header"],
}
)
def test_check_access(
self,
access_type,
inet_exposure,
local_ranges,
remote_ip,
xff_header,
verify_xff_header,
result_with_xff,
monkeypatch,
):
def _func():
request = proxied_request(remote_ip, xff_header=xff_header)
if not verify_xff_header and xff_header:
# Nothing is trusted, so a header leaves us with no client to grant access to
result = False
elif verify_xff_header:
result = result_with_xff
else:
# Without XFF, only the remote IP and the local ranges setting matter
result = is_loopback_addr(remote_ip) or is_local_addr(remote_ip)
if access_type <= inet_exposure:
assert security.check_access(request, access_type) is True
else:
assert security.check_access(request, access_type) is result
_func()
@pytest.mark.config({"api_key": "the_real_api_key", "nzb_key": "the_real_nzb_key", "api_warnings": True})
@pytest.mark.parametrize(
"params, expected",
[
({"mode": "version"}, None),
({"mode": "auth"}, None),
# The NZB-key is valid for nzb-level calls only
({"mode": "addfile", "apikey": "the_real_nzb_key"}, None),
({"mode": "queue", "apikey": "the_real_nzb_key"}, interface._MSG_APIKEY_INCORRECT),
({"mode": "queue", "apikey": "the_real_api_key"}, None),
({"mode": "queue"}, interface._MSG_APIKEY_REQUIRED),
({"mode": "queue", "apikey": "wrong"}, interface._MSG_APIKEY_INCORRECT),
],
)
def test_apikey_on_the_api_route(self, params, expected):
request = mock_request(params=params)
response = interface.check_apikey(request)
assert (response.body.decode() if response else None) == expected
@pytest.mark.parametrize(
"local_ranges, xff_ips, expected_result",
[
([], ["4.3.2.1"], "4.3.2.1"), # Standard situation, single non-local XFF IP
([], ["42:1b5::beef"], "42:1b5::beef"),
([], ["10.10.10.10"], "10.10.10.10"), # Only local XFF IPs, first entry wins
([], ["::1"], "::1"),
([], ["127.0.0.1"], "127.0.0.1"),
([], ["10.10.10.10", "192.168.0.1"], "10.10.10.10"), # Only local XFF IPs, first entry wins
([], ["10.10.10.10", "192.168.0.1", "192.168.1.2"], "10.10.10.10"), # Only local XFF IPs, first entry wins
([], ["4.3.2.1", "10.10.10.10"], "4.3.2.1"), # First non-local entry wins
([], ["4.3.2.1", "192.168.0.1", "10.10.10.10"], "4.3.2.1"),
([], ["127.0.0.1", "4.3.2.1", "192.168.0.1", "10.10.10.10"], "4.3.2.1"),
([], ["4.3.2.1", "192.168.0.1", "10.10.10.10", "127.0.0.1"], "4.3.2.1"),
([], ["666::1", "4.3.2.1", "10.10.10.10"], "4.3.2.1"),
([], ["4.3.2.1", "666::1", "192.168.0.1", "10.10.10.10"], "666::1"),
([], ["127.0.0.1", "4.3.2.1", "666::1", "10.10.10.10"], "666::1"),
([], ["4.3.2.1", "192.168.0.1", "10.10.10.10", "127.0.0.1", "666::1"], "666::1"),
([], ["10.10.10.10", "4.3.2.1"], "4.3.2.1"),
([], ["192.168.0.1", "4.3.2.1", "10.10.10.10"], "4.3.2.1"),
([], ["127.0.0.1", "192.168.0.1", "4.3.2.1", "10.10.10.10"], "4.3.2.1"),
([], ["192.168.0.1", "4.3.2.1", "10.10.10.10", "127.0.0.1"], "4.3.2.1"),
(["4.3.2.0/24"], ["4.3.2.1"], "4.3.2.1"), # Only local IPs due to local_ranges, first entry wins
(["666::/48"], ["666::1"], "666::1"),
(["192.168.0.0/16", "4.3.2.0/24"], ["4.3.2.1"], "4.3.2.1"),
(["666::/48", "192.168.0.0/16", "4.3.2.0/24"], ["4.3.2.1"], "4.3.2.1"),
(["666::/48", "192.168.0.0/16", "4.3.2.0/24"], ["666::1"], "666::1"),
(["192.168.0.0/16", "4.3.2.0/24"], ["10.10.10.10"], "10.10.10.10"), # 10.x wins, outside local_ranges
(["192.168.0.0/16", "4.3.2.0/24"], ["4.3.2.1", "10.10.10.10"], "10.10.10.10"),
(
["192.168.0.0/16", "4.3.2.0/24"],
["10.10.10.10", "4.3.2.1"],
"4.3.2.1",
), # Public local range, so no longer a trusted hop
(["666::/48", "192.168.0.0/16", "4.3.2.0/24"], ["10.10.10.10"], "10.10.10.10"),
(["666::/48", "192.168.0.0/16", "4.3.2.0/24"], ["4.3.2.1", "10.10.10.10"], "10.10.10.10"),
(
["666::/48", "192.168.0.0/16", "4.3.2.0/24"],
["10.10.10.10", "4.3.2.1"],
"4.3.2.1",
), # Public local range, so no longer a trusted hop
(["8.8.8.8", "4.3.2.1"], ["4.3.2.1", "192.168.0.1", "10.10.10.10"], "10.10.10.10"),
(["8.8.8.8", "4.3.2.1"], ["192.168.0.1", "4.3.2.1", "10.10.10.10"], "10.10.10.10"),
(
["8.8.8.8", "4.3.2.1"],
["192.168.0.1", "10.10.10.10", "4.3.2.1"],
"4.3.2.1",
), # Public local range, so no longer a trusted hop
(["8.8.8.8"], ["192.168.0.1", "10.10.10.10", "4.3.2.1"], "4.3.2.1"), # All XFF IPs non-local, last wins
(["8.8.8.8"], ["4.3.2.1", "10.10.10.10", "192.168.0.1"], "192.168.0.1"),
(["8.8.8.8"], ["4.3.2.1", "192.168.0.1", "10.10.10.10"], "10.10.10.10"),
(["8.8.8.8"], ["127.0.0.1", "4.3.2.1", "10.10.10.10", "192.168.0.1"], "192.168.0.1"),
(["666::/48"], ["192.168.0.1", "10.10.10.10", "4.3.2.1"], "4.3.2.1"),
(["666::/48"], ["4.3.2.1", "10.10.10.10", "192.168.0.1"], "192.168.0.1"),
(["666::/48"], ["4.3.2.1", "192.168.0.1", "10.10.10.10"], "10.10.10.10"),
(["666::/48"], ["127.0.0.1", "4.3.2.1", "10.10.10.10", "192.168.0.1"], "192.168.0.1"),
(["8.8.8.8"], ["4.3.2.1", "192.168.0.1", "10.10.10.10", "127.0.0.1"], "10.10.10.10"), # Loopback as last
(["666::/48"], ["4.3.2.1", "192.168.0.1", "10.10.10.10", "127.0.0.1"], "10.10.10.10"),
(["8.8.8.8"], ["4.3.2.1", "192.168.0.1", "10.10.10.10", "::1"], "10.10.10.10"),
(["666::/48"], ["4.3.2.1", "192.168.0.1", "10.10.10.10", "::1"], "10.10.10.10"),
([], ["4.3.2.1:56789"], "4.3.2.1"), # Port stripped from XFF entry
],
)
@pytest.mark.config(
lambda params: {
"local_ranges": params["local_ranges"],
}
)
def test_effective_client_from_xff(self, local_ranges, xff_ips, expected_result):
def _func():
# The effective client IP (used for access
# checks) is selected by uvicorn's ProxyHeadersMiddleware: the last
# XFF entry that is not a trusted (local) proxy, or the first entry
# when the whole chain is trusted. Connect from loopback, which is
# always a trusted peer.
assert xff_ips
client = resolve_client(remote_ip="127.0.0.1", xff_header=", ".join(xff_ips))
assert client.host == expected_result
_func()
@pytest.mark.parametrize("access_type", [1, 2, 3, 4, 5, 6])
@pytest.mark.parametrize("inet_exposure", [0, 2, 4])
@pytest.mark.config(lambda params: {"inet_exposure": params["inet_exposure"], "api_warnings": True})
def test_check_access_without_client(self, access_type, inet_exposure):
# request.client can be None on unix sockets and with some test clients
request = mock_request()
request.client = None
assert security.check_access(request, access_type, warn_user=True) is (access_type <= inet_exposure)
# The logging helpers must not raise either
interface.log_warning_and_ip(request, "txt")
@pytest.mark.parametrize(
"local_ranges, expected_networks, unexpected_networks",
[
# Without local_ranges: loopback plus all private address space
(None, ["127.0.0.0/8", "::1/128", "10.0.0.0/8", "192.168.0.0/16", "::ffff:10.0.0.0/104"], []),
# With local_ranges: loopback plus the configured ranges only
(
"192.168.1.0/24",
["127.0.0.0/8", "::1/128", "192.168.1.0/24", "::ffff:192.168.1.0/120"],
["10.0.0.0/8", "172.16.0.0/12"],
),
],
)
@pytest.mark.config(lambda params: {"local_ranges": params["local_ranges"]})
def test_xff_trusted_networks(self, local_ranges, expected_networks, unexpected_networks):
def _func():
networks = xff_trusted_networks()
for network in expected_networks:
assert network in networks
for network in unexpected_networks:
assert network not in networks
_func()
async def empty_app(scope, receive, send):
"""Minimal ASGI app, replying to anything that does reach it"""
await send({"type": "http.response.start", "status": 200, "headers": []})
await send({"type": "http.response.body", "body": b""})
def is_client_error(record: logging.LogRecord) -> bool:
"""Was this record logged by uvicorn because of client behavior?"""
return record.getMessage().startswith(interface.UvicornNoiseFilter.CLIENT_ERRORS)
def feed_raw_request(protocol_class, data: bytes):
"""Hand raw bytes to a real uvicorn HTTP protocol, so it logs whatever it
makes of them, just like it would for an actual connection."""
async def _run():
# log_config=None: the logging setup under test is applied by the fixture
config = uvicorn.Config(app=empty_app, log_config=None)
config.load()
protocol = protocol_class(config=config, server_state=ServerState(), app_state={})
transport = Mock()
transport.get_extra_info = lambda name, default=None: ("127.0.0.1", 12345) if name == "peername" else default
protocol.connection_made(transport)
protocol.data_received(data)
run_async(_run())
class TestUvicornLogging:
@pytest.fixture(autouse=True)
def uvicorn_logging(self):
"""Apply the logging configuration that SABnzbd hands to uvicorn and
restore the previous state afterwards, so other tests are unaffected."""
loggers = [logging.getLogger(name) for name in ("uvicorn", "uvicorn.error", "uvicorn.access")]
saved = [(logger.level, logger.propagate, logger.handlers[:], logger.filters[:]) for logger in loggers]
logging.config.dictConfig(interface.uvicorn_logging_config())
yield
for logger, (level, propagate, handlers, filters) in zip(loggers, saved):
logger.setLevel(level)
logger.propagate = propagate
logger.handlers = handlers
logger.filters = filters
@pytest.mark.parametrize("protocol_class", [h11_impl.H11Protocol, httptools_impl.HttpToolsProtocol])
@pytest.mark.parametrize(
"raw_request",
[
# Not HTTP at all, as sent by port scanners and misdirected clients
b"\x16\x03\x01\x00\xf4\x01\x00\x00\xf0\x03\x03",
# Valid HTTP, but asking for an upgrade we do not support
b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: upgrade\r\nUpgrade: h2c\r\n\r\n",
],
)
def test_client_errors_are_not_warnings(self, protocol_class, raw_request, caplog):
# A handler that only wants INFO and up, just like the console and logfile
# handlers when debug logging is off, must not see the message at all
caplog.set_level(logging.INFO)
feed_raw_request(protocol_class, raw_request)
assert not [record for record in caplog.records if record.levelno >= logging.WARNING]
assert not [record for record in caplog.records if is_client_error(record)]
@pytest.mark.parametrize("protocol_class", [h11_impl.H11Protocol, httptools_impl.HttpToolsProtocol])
def test_client_errors_are_kept_for_debug_logging(self, protocol_class, caplog):
caplog.set_level(logging.DEBUG)
feed_raw_request(protocol_class, b"\x16\x03\x01\x00\xf4\x01\x00\x00\xf0\x03\x03")
assert [
record
for record in caplog.records
if record.levelname == "DEBUG" and record.getMessage() == "Invalid HTTP request received."
]
def test_lifecycle_messages_are_not_logged(self, caplog):
# Starting and stopping is already logged by SABnzbd itself
caplog.set_level(logging.INFO)
logging.getLogger("uvicorn.error").info("Application startup complete.")
assert not caplog.records
def test_lifecycle_messages_are_kept_for_debug_logging(self, caplog):
caplog.set_level(logging.DEBUG)
logging.getLogger("uvicorn.error").info("Application startup complete.")
assert [record for record in caplog.records if record.levelname == "DEBUG"]
def test_failures_reported_by_sabnzbd_are_not_logged_twice(self, caplog):
# SABnzbd logs its own error, including the reason, when the
# web-interface fails to start, so this summary adds nothing
caplog.set_level(logging.INFO)
logging.getLogger("uvicorn.error").error("Application startup failed. Exiting.")
assert not caplog.records
def test_reason_for_a_failed_start_still_propagates(self, caplog):
caplog.set_level(logging.INFO)
logging.getLogger("uvicorn.error").error("Exception in 'lifespan' protocol")
assert [record for record in caplog.records if record.levelno == logging.ERROR]
def test_real_warnings_still_propagate(self, caplog):
caplog.set_level(logging.INFO)
logging.getLogger("uvicorn.error").warning("Exceeded concurrency limit.")
logging.getLogger("uvicorn.error").error("Exception in ASGI application")
assert [record for record in caplog.records if record.levelno == logging.WARNING]
assert [record for record in caplog.records if record.levelno == logging.ERROR]
def test_filtered_messages_still_used_by_uvicorn(self):
"""Guard against uvicorn rewording the messages we filter on"""
uvicorn_source = "".join(inspect.getsource(module) for module in (h11_impl, httptools_impl, lifespan_on))
for message in interface.UvicornNoiseFilter.CLIENT_ERRORS + interface.UvicornNoiseFilter.REPORTED_FAILURES:
assert message in uvicorn_source
class TestClientAddressInfo:
"""The client address goes into log lines as host:port, so an IPv6 address has to
be bracketed: ::ffff:127.0.0.1:55170 gives no clue where the address stops."""
@pytest.mark.config({"verify_xff_header": False})
@pytest.mark.parametrize(
"remote_ip, expected",
[
("127.0.0.1", "127.0.0.1:55170 [-]"),
("10.11.12.13", "10.11.12.13:55170 [-]"),
("::1", "[::1]:55170 [-]"),
# Dual-stack listener reporting an IPv4 client
("::ffff:127.0.0.1", "[::ffff:127.0.0.1]:55170 [-]"),
("2001:470:1:332::152", "[2001:470:1:332::152]:55170 [-]"),
# Unknown client, request.client was None
("", ":55170 [-]"),
],
)
def test_brackets_ipv6(self, remote_ip, expected):
request = mock_request(remote_ip=remote_ip, remote_port=55170)
assert security.client_address_info(request) == expected
@pytest.mark.parametrize("user_agent", ["curl/8.0", None, ""])
@pytest.mark.config({"verify_xff_header": False})
def test_always_ends_with_a_user_agent(self, user_agent):
"""sanitize_line() only recognises a remote label that ends in a bracketed
user-agent, and a label it does not recognise is not redacted at all. The brackets
therefore have to hold something even when the request sent no user-agent."""
headers = {"User-Agent": user_agent} if user_agent is not None else None
request = mock_request(remote_ip="8.8.8.8", remote_port=55170, headers=headers)
info = security.client_address_info(request)
assert info.endswith(f" [{user_agent or '-'}]")
assert api.sanitize_line(f"Refused connection from: {info}".encode()).count(b"<REMOVED>") == 1
@pytest.mark.config({"verify_xff_header": True})
def test_includes_forwarded_chain(self):
request = mock_request(remote_ip="::1", remote_port=55170, headers={"X-Forwarded-For": "8.7.6.5, ::1"})
assert security.client_address_info(request) == "[::1]:55170 (X-Forwarded-For: 8.7.6.5, ::1) [-]"
@pytest.mark.config({"verify_xff_header": False})
def test_omits_forwarded_chain_when_not_verified(self):
"""Without verify_xff_header the header is not trusted, so it is not reported"""
request = mock_request(remote_ip="::1", remote_port=55170, headers={"X-Forwarded-For": "8.7.6.5"})
assert security.client_address_info(request) == "[::1]:55170 [-]"
class TestUseSecureCookies:
"""The Secure attribute must follow the connection the request actually arrived on,
including TLS terminated by a trusted reverse proxy in front of SABnzbd."""
@staticmethod
def make_request(scheme: str, host: str | None = "sab.example.com", server=("127.0.0.1", 8080)) -> Request:
headers = [(b"host", host.encode())] if host is not None else []
return Request(
{
"type": "http",
"method": "GET",
"path": "/",
"query_string": b"",
"scheme": scheme,
"headers": headers,
"client": ("127.0.0.1", 12345),
"server": server,
}
)
@pytest.mark.config({"enable_https": False})
@pytest.mark.parametrize(
"scheme, host, server, expected",
[
("http", "sab.example.com", ("127.0.0.1", 8080), False),
("https", "sab.example.com", ("127.0.0.1", 8080), True),
# No Host header: the scheme must still decide the flag
("https", None, ("127.0.0.1", 8080), True),
# An IPv6 listen address leaves the URL unparseable, the scheme is unaffected
("https", None, ("::ffff:127.0.0.1", 8080), True),
("https", "1234:5678::1:8080", ("::1", 8080), True),
# Neither a Host header nor an address: the URL is relative and has no
# scheme at all, which must not silently drop the Secure attribute
("https", None, None, True),
("http", None, None, False),
],
)
def test_follows_request_scheme(self, scheme, host, server, expected):
assert security.use_secure_cookies(self.make_request(scheme, host, server)) is expected
@pytest.mark.config({"enable_https": True})
def test_https_enabled_always_secure(self):
"""Serving https ourselves is enough, whatever the request looks like"""
assert security.use_secure_cookies(self.make_request("http")) is True
@pytest.mark.config({"enable_https": False})
def test_scheme_from_trusted_proxy(self):
"""X-Forwarded-Proto from a trusted proxy is resolved into the scope by
uvicorn, so a proxy terminating TLS still gets the Secure attribute."""
captured = {}
async def asgi_app(scope, receive, send):
captured["secure"] = security.use_secure_cookies(Request(scope))
def run(remote_ip: str):
middleware = ProxyHeadersMiddleware(asgi_app, trusted_hosts=xff_trusted_networks())
scope = {
"type": "http",
"method": "GET",
"path": "/",
"query_string": b"",
"scheme": "http",
"client": (remote_ip, 12345),
"server": ("127.0.0.1", 8080),
"headers": [(b"host", b"sab.example.com"), (b"x-forwarded-proto", b"https")],
}
run_async(middleware(scope, None, None))
return captured["secure"]
# Trusted proxy: the forwarded scheme is honoured
assert run("127.0.0.1") is True
# Untrusted peer: the header must be ignored, so no Secure on a plain connection
assert run("8.7.6.5") is False
class TestStaleTokenIsNotAWarning:
def _levels(self, caplog, run) -> set:
caplog.clear()
with caplog.at_level(logging.INFO, logger=""):
run()
return {record.levelname for record in caplog.records}
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0, "api_warnings": True})
def test_api_call_with_a_stale_token_is_only_info(self, session_store, caplog):
request = mock_request(params={"mode": "queue", "name": ""}, csrf="f0" * 32)
levels = self._levels(caplog, lambda: interface.check_apikey(request))
assert "WARNING" not in levels
assert "INFO" in levels
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0, "api_warnings": True})
def test_api_call_with_no_token_still_warns(self, session_store, caplog):
request = api_request(None, with_token=False)
levels = self._levels(caplog, lambda: interface.check_apikey(request))
assert "WARNING" in levels
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0, "api_warnings": True})
def test_page_post_with_a_stale_token_is_only_info(self, session_store, caplog):
request = page_post(csrf="f0" * 32)
levels = self._levels(caplog, lambda: config_save_middleware().denied_response(request))
assert "WARNING" not in levels
assert "INFO" in levels
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0, "api_warnings": True})
def test_cross_site_page_post_still_warns(self, session_store, caplog):
levels = self._levels(caplog, lambda: config_save_middleware().denied_response(page_post()))
assert "WARNING" in levels
class TestApiCsrf:
"""A cookie-authorized API call must echo its session's CSRF token in a header; with the
login bypassed the token stands alone, bound to a stable identity and carrying no cookie"""
def _status(self, request) -> Optional[int]:
response = interface.check_apikey(request)
return response.status_code if response else None
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_token_without_cookie_authorizes(self, session_store):
assert self._status(api_request(None)) is None
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_missing_token_is_forbidden(self, session_store):
"""403, not 401: a reload cannot conjure up a header the client never sends"""
assert self._status(api_request(None, with_token=False)) == 403
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_stale_token_asks_for_a_reload(self, session_store):
"""401, because a reload does fix this one: a page from before a key rotation"""
request = mock_request(params={"mode": "queue", "name": ""}, csrf="f0" * 32)
assert self._status(request) == 401
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_apikey_still_works_alongside_a_cookie(self, session_store):
"""The cookie check must fall through to the key, never reject on its own"""
request = mock_request("f0" * 32, params={"mode": "queue", "name": "", "apikey": cfg.api_key()})
assert self._status(request) is None
# ...and with no cookie at all, which is how 3rd-party clients call it
assert self._status(mock_request(params={"mode": "queue", "name": "", "apikey": cfg.api_key()})) is None
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_no_mode_is_exempt(self, session_store):
"""There is no carve-out list, including showlog"""
assert self._status(api_request(None, mode="showlog", with_token=False)) == 403
assert self._status(api_request(None, mode="showlog")) is None
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_token_outside_the_header_is_not_accepted(self, session_store):
"""This route merges the query string into its params; page routes still take the field"""
as_parameter = mock_request(params={"mode": "queue", "name": "", security.CSRF_FIELD: anonymous_csrf()})
assert self._status(as_parameter) == 403
# The same token in the header is what the frontend sends, and it is accepted
assert self._status(api_request(None)) is None
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_keyless_and_cookieless_still_reports_missing_key(self, session_store):
assert self._status(mock_request(params={"mode": "queue", "name": ""})) == 403
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_apikey_is_compared_in_constant_time(self, session_store):
key = cfg.api_key()
assert self._status(mock_request(params={"mode": "queue", "name": "", "apikey": key})) is None
# A prefix of the real key is no better than any other wrong key
for wrong in (key[:-1], key[:1], "", "x" * len(key)):
request = mock_request(params={"mode": "queue", "name": "", "apikey": wrong})
assert self._status(request) == 403
@pytest.mark.parametrize("credentials", [("", ""), ("user", "pass")])
@pytest.mark.config(
lambda params: {
"username": params["credentials"][0],
"password": params["credentials"][1],
"inet_exposure": 0,
"api_warnings": True,
}
)
def test_apikey_does_not_open_the_page_routes(self, session_store, credentials):
"""Refused the same way whether or not credentials are configured, and never with a redirect to the login form"""
request = page_post()
request.state.params = {"apikey": cfg.api_key()}
request.query_params = QueryParams("")
response = config_save_middleware().denied_response(request)
assert response is not None
assert response.status_code == 403
assert b"only accepted on /api" in response.body
@pytest.mark.config({"username": "user", "password": "pass", "inet_exposure": 0})
def test_stray_apikey_does_not_break_a_valid_session(self, session_store):
store_session(session_store, "login-token")
request = page_post("login-token", csrf=security.csrf_token_for("login-token"))
request.state.params = {security.CSRF_FIELD: security.csrf_token_for("login-token")}
request.query_params = QueryParams("apikey=" + cfg.api_key())
assert config_save_middleware().denied_response(request) is None
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_version_and_auth_skip_the_check(self, session_store):
for mode in ("version", "auth"):
assert self._status(mock_request(params={"mode": mode, "name": ""})) is None
class TestLogRoute:
def _route(self):
return next(route for route in interface.INTERFACE_ROUTES if getattr(route, "path", None) == "/log")
def test_registered_as_a_guarded_post_route(self):
assert sorted(self._route().methods) == ["POST"]
@pytest.mark.config({"username": "user", "password": "pass", "inet_exposure": 0})
def test_behind_the_login_check(self, session_store):
"""Driven through the route as registered, so weakening the decorator fails here"""
route = self._route()
captured = {}
async def receive():
return {"type": "http.request", "body": b"", "more_body": False}
async def send(message):
if message["type"] == "http.response.start":
captured["status"] = message["status"]
captured["headers"] = Headers(raw=message["headers"])
scope = {
"type": "http",
"method": "POST",
"path": "/log",
"query_string": b"",
"headers": [(b"host", b"127.0.0.1:8080")],
"client": ("127.0.0.1", 12345),
"server": ("127.0.0.1", 8080),
"scheme": "http",
}
run_async(route.app(scope, receive, send))
# Redirected to the login form, and the handler never ran to stream any of the log
assert captured["status"] == 302
assert captured["headers"]["location"].endswith("/login")
# The kinds of session cookie a page POST can arrive with, resolved once the test's
# credentials are in place
COOKIE_NONE = "none"
COOKIE_LOGIN = "login"
COOKIE_FORGED = "forged"
def cookie_of_kind(kind: str, store) -> Optional[str]:
if kind == COOKIE_NONE:
return None
if kind == COOKIE_FORGED:
return "f0" * 32
store_session(store, "login-token")
return "login-token"
# The CSRF token a page POST can arrive with, relative to the identity it presents
TOKEN_NONE = "none"
TOKEN_MATCHING = "matching"
TOKEN_WRONG = "wrong"
def apply_token_of_kind(kind: str, request):
"""Set the request's CSRF header to the given kind, relative to its own identity"""
if kind == TOKEN_MATCHING:
set_csrf_header(request, security.csrf_token_for(security.csrf_identity(request)))
elif kind == TOKEN_WRONG:
set_csrf_header(request, "f0" * 32)
class TestPagePostCsrf:
@pytest.mark.parametrize(
"credentials, inet_exposure, cookie, token, allowed",
[
# Login bypassed: there is no cookie, so the CSRF token alone gates the POST
(("", ""), 0, COOKIE_NONE, TOKEN_NONE, False),
(("", ""), 0, COOKIE_NONE, TOKEN_MATCHING, True),
(("", ""), 0, COOKIE_NONE, TOKEN_WRONG, False),
(("", ""), 5, COOKIE_NONE, TOKEN_NONE, False),
# Credentials with the login enforced: a valid session cookie is required
(("user", "pass"), 0, COOKIE_NONE, TOKEN_NONE, False),
# A matching token with no session cookie still fails the login check
(("user", "pass"), 0, COOKIE_NONE, TOKEN_MATCHING, False),
(("user", "pass"), 0, COOKIE_LOGIN, TOKEN_MATCHING, True),
(("user", "pass"), 0, COOKIE_LOGIN, TOKEN_NONE, False),
# A forged cookie is no session, even with the token that matches its identity
(("user", "pass"), 0, COOKIE_FORGED, TOKEN_MATCHING, False),
# inet_exposure 5: the login is waived for local clients, so no cookie is needed
(("user", "pass"), 5, COOKIE_NONE, TOKEN_NONE, False),
(("user", "pass"), 5, COOKIE_NONE, TOKEN_MATCHING, True),
# ...without locking out a local client who does hold a real login session
(("user", "pass"), 5, COOKIE_LOGIN, TOKEN_MATCHING, True),
],
)
@pytest.mark.config(
lambda params: {
"username": params["credentials"][0],
"password": params["credentials"][1],
"inet_exposure": params["inet_exposure"],
}
)
def test_config_save_post(self, session_store, credentials, inet_exposure, cookie, token, allowed):
request = page_post(cookie_of_kind(cookie, session_store))
apply_token_of_kind(token, request)
response = config_save_middleware().denied_response(request)
assert (response is None) is allowed
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_token_accepted_as_form_field(self, session_store):
"""A form that navigates cannot set a header, so the body is accepted too"""
allowed = page_post(None, csrf_field=anonymous_csrf())
assert config_save_middleware().denied_response(allowed) is None
denied = page_post(None, csrf_field="f0" * 32)
assert config_save_middleware().denied_response(denied) is not None
@pytest.mark.config({"username": "user", "password": "pass", "inet_exposure": 5})
def test_token_is_bound_to_its_own_session(self, session_store):
store_session(session_store, "login-token")
# A login session's cookie carrying the anonymous token (the wrong identity) is refused
login_with_anonymous_token = page_post("login-token", csrf=anonymous_csrf())
assert config_save_middleware().denied_response(login_with_anonymous_token) is not None
# And the login token presented with no cookie (the anonymous identity) is refused
no_cookie_with_login_token = page_post(None, csrf=security.csrf_token_for("login-token"))
assert config_save_middleware().denied_response(no_cookie_with_login_token) is not None
@pytest.mark.config({"username": "user", "password": "pass", "inet_exposure": 5})
def test_external_client_still_needs_login(self, session_store):
"""The waiver is local-only: an external POST gets no help from the CSRF token"""
request = page_post(None, remote_ip="9.8.7.6", csrf=anonymous_csrf())
assert config_save_middleware().denied_response(request) is not None
@pytest.mark.config({"username": "user", "password": "pass", "inet_exposure": 0})
def test_session_is_looked_up_once_per_request(self, session_store):
store_session(session_store, "login-token")
request = page_post("login-token", csrf=security.csrf_token_for("login-token"))
with patch.object(sabnzbd.SessionStore, "get", wraps=sabnzbd.SessionStore.get) as get_session:
assert config_save_middleware().denied_response(request) is None
assert get_session.call_count == 1
@pytest.mark.config({"username": "", "password": "", "inet_exposure": 0})
def test_get_is_not_guarded(self, session_store):
"""A first visit has no token yet, so rendering a page must not require one"""
assert config_save_middleware().denied_response(mock_request()) is None
def run_page_request(cookie: Optional[str] = None, remote_ip: str = "127.0.0.1") -> tuple[list[str], str]:
"""Drive a page route's SecurityMiddleware over a real ASGI scope, returning the Set-Cookie values it injected and the CSRF token it published"""
rendered_token = ""
async def asgi_app(scope, receive, send):
# Stand in for a page handler: build_header reads the token off request.state here
nonlocal rendered_token
rendered_token = scope.get("state", {}).get("csrf_token", "")
await send({"type": "http.response.start", "status": 200, "headers": []})
await send({"type": "http.response.body", "body": b"ok"})
middleware = interface.SecurityMiddleware(asgi_app, check_for_login=True, check_api_key=False, access_type=4)
headers = [(b"host", b"127.0.0.1:8080")]
if cookie:
headers.append((b"cookie", ("%s=%s" % (security.SESSION_COOKIE_USER, cookie)).encode()))
scope = {
"type": "http",
"method": "GET",
"path": "/config/general",
"query_string": b"",
"headers": headers,
"client": (remote_ip, 12345),
"server": ("127.0.0.1", 8080),
"scheme": "http",
# secured_expose always wraps SecurityMiddleware in ParamsMiddleware
"state": {"params": QueryParams("")},
}
captured: list[str] = []
async def send(message):
if message["type"] == "http.response.start":
captured.extend(Headers(raw=message["headers"]).getlist("set-cookie"))
run_async(middleware(scope, None, send))
return captured, rendered_token
class TestPageLoadIssuesNoSessionCookie:
"""Page loads never set a session cookie; login sessions come only from the login form"""
def _session_cookies(self, **kwargs) -> list[str]:
set_cookies, _ = run_page_request(**kwargs)
return [value for value in set_cookies if value.startswith(security.SESSION_COOKIE_USER + "=")]
@pytest.mark.config({"username": "", "password": ""})
def test_none_issued_when_login_bypassed(self, session_store):
assert self._session_cookies() == []
@pytest.mark.config({"username": "user", "password": "pass", "inet_exposure": 5})
def test_none_issued_for_waived_local_client(self, session_store):
assert self._session_cookies() == []
@pytest.mark.config({"username": "user", "password": "pass", "inet_exposure": 5})
def test_login_session_left_untouched(self, session_store):
store_session(session_store, "login-token")
assert self._session_cookies(cookie="login-token") == []
class TestRenderedToken:
"""The token a page renders is the one that authorizes its next POST"""
def _rendered(self, cookie: Optional[str] = None, remote_ip: str = "127.0.0.1") -> str:
_set_cookies, rendered_token = run_page_request(cookie=cookie, remote_ip=remote_ip)
assert rendered_token, "a page rendered no usable token at all"
return rendered_token
@pytest.mark.config({"username": "", "password": ""})
def test_bypassed_binds_to_the_stable_identity(self, session_store):
# No cookie is needed: the token binds to the stable anonymous identity
assert self._rendered(None) == anonymous_csrf()
@pytest.mark.config({"username": "user", "password": "pass", "inet_exposure": 0})
def test_login_session_binds_to_its_cookie(self, session_store):
store_session(session_store, "login-token")
assert self._rendered("login-token") == security.csrf_token_for("login-token")
@pytest.mark.config({"username": "user", "password": "pass", "inet_exposure": 5})
def test_login_session_while_login_waived(self, session_store):
store_session(session_store, "login-token")
assert self._rendered("login-token") == security.csrf_token_for("login-token")
@pytest.mark.config({"username": "", "password": ""})
def test_rendered_token_authorizes_the_next_post(self, session_store):
rendered = self._rendered(None)
request = page_post(None, csrf=rendered)
assert config_save_middleware().denied_response(request) is None
Reference in new issue View Git Blame Copy Permalink