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

805 lines
36 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_api - Tests for API functions
"""
import os
from functools import cached_property
import pytest
from random import choice, randint
from unittest import mock
from unittest.mock import Mock, patch
from starlette.requests import Request
from starlette.responses import Response
from starlette.datastructures import Headers, Address, QueryParams, State
import sabnzbd.api as api
import sabnzbd.interface as interface
import sabnzbd.security as security
import sabnzbd
import sabnzbd.database as db
from sabnzbd.constants import DB_HISTORY_NAME, DEF_ADMIN_DIR, PP_LOOKUP, AddNzbFileResult, Status
from sabnzbd.misc import pp_to_opts
from tests.testhelper import FakeHistoryDB, SAB_CACHE_DIR, run_async
from tests.test_interface import resolve_client
def run_api_handler(kwargs) -> Response:
"""Run the (async) api_handler to completion, like the /api route does"""
return run_async(api.api_handler(kwargs))
class TestApiInternals:
"""Test internal functions of the API"""
def test_empty(self):
with pytest.raises(AttributeError):
run_api_handler(None)
# Empty string should work but result in undefined mode
result = run_api_handler(QueryParams({}))
assert "not implemented" in result.body.decode()
def test_mode_invalid(self):
result = run_api_handler(QueryParams({"mode": "invalid"}))
assert "not implemented" in result.body.decode()
def test_version(self):
result = run_api_handler(QueryParams({"mode": "version"}))
assert sabnzbd.__version__ in result.body.decode()
def test_auth(self):
result = run_api_handler(QueryParams({"mode": "auth"}))
assert "apikey" in result.body.decode()
@pytest.mark.parametrize(
"line,ip",
[
(
b"2026-05-19 18:35:18,271::INFO::[notifier:169] Sending notification: Warning - Unsuccessful login attempt from ::ffff:172.18.0.1 [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0] (type=warning, job_cat=None)\n",
b"::ffff:172.18.0.1",
),
(
b"2026-05-19 18:35:18,271::WARNING::[interface:689] Unsuccessful login attempt from 172.18.0.1 [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]\n",
b"172.18.0.1",
),
(
b"2026-05-19 18:35:18,271::WARNING::[interface:689] Unsuccessful login attempt from 2001:4860::1 [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]\n",
b"<REMOVED>",
),
(
b"2026-05-19 18:35:18,271::WARNING::[interface:689] Unsuccessful login attempt from 8.8.8.8 [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]\n",
b"<REMOVED>",
),
(
b"2026-05-19 18:35:18,271::WARNING::[interface:689] Unsuccessful login attempt from 127.0.0.1 [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]\n",
b"127.0.0.1",
),
(
b"2026-05-19 18:35:18,271::WARNING::[interface:689] Unsuccessful login attempt from fe80::1 [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]\n",
b"fe80::1",
),
],
ids=[
"ipv6-local",
"ipv4-local",
"ipv6-removed",
"ipv4-removed",
"ipv4-loopback",
"ipv6-linklocal",
],
)
def test_log_sanitize_remote_label(self, line, ip):
sanitized = api.sanitize_line(line)
assert (
sanitized.count(
b"%s [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]" % ip
)
== 1
)
@pytest.mark.parametrize(
"line,ip,xff",
[
(
b"2026-05-19 18:35:18,271::INFO::[notifier:169] Sending notification: Warning - Unsuccessful login attempt from ::ffff:172.18.0.1 (X-Forwarded-For: 8.8.8.8, 1.1.1.1) [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0] (type=warning, job_cat=None)\n",
b"::ffff:172.18.0.1",
[b"<REMOVED>", b"<REMOVED>"],
),
(
b"2026-05-19 18:35:18,271::WARNING::[interface:689] Unsuccessful login attempt from 172.18.0.1 (X-Forwarded-For: 8.8.8.8, 1.1.1.1) [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]\n",
b"172.18.0.1",
[b"<REMOVED>", b"<REMOVED>"],
),
(
b"2026-05-19 18:35:18,271::WARNING::[interface:689] Unsuccessful login attempt from 2001:4860::1 (X-Forwarded-For: 192.168.0.50, 8.8.8.8) [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]\n",
b"<REMOVED>",
[b"192.168.0.50", b"<REMOVED>"],
),
(
b"2026-05-19 18:35:18,271::WARNING::[interface:689] Unsuccessful login attempt from 8.8.8.8 (X-Forwarded-For: 1.1.1.1) [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]\n",
b"<REMOVED>",
[b"<REMOVED>"],
),
(
b"2026-05-19 18:35:18,271::WARNING::[interface:689] Unsuccessful login attempt from 127.0.0.1 (X-Forwarded-For: 127.1.2.3) [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]\n",
b"127.0.0.1",
[b"127.1.2.3"],
),
(
b"2026-05-19 18:35:18,271::WARNING::[interface:689] Unsuccessful login attempt from fe80::1 (X-Forwarded-For: fe80::2) [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]\n",
b"fe80::1",
[b"fe80::2"],
),
],
ids=[
"ipv6-local-removed-removed",
"ipv4-local-removed-removed",
"ipv6-removed-local-removed",
"ipv4-removed-removed",
"ipv4-loopback-loopback",
"ipv6-linklocal-linklocal",
],
)
def test_log_sanitize_remote_label_xff(self, line, ip, xff):
sanitized = api.sanitize_line(line)
assert (
sanitized.count(
b"%s (X-Forwarded-For: %s) [Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0]"
% (ip, b", ".join(xff))
)
== 1
)
@pytest.mark.parametrize(
"label, expected",
[
# host:port, which the pattern has to allow for or "8.8.8.8:12345" reads as
# the IPv6-looking "8:12345"
(b"8.8.8.8:12345", b"<REMOVED>:12345"),
(b"192.168.1.5:12345", b"192.168.1.5:12345"),
# An IPv6 client is bracketed, and the brackets are part of the label
(b"[2001:4860::1]:12345", b"<REMOVED>:12345"),
(b"[fe80::1]:12345", b"[fe80::1]:12345"),
# The peer a request arrived through is redacted like any other address
(b"192.168.1.5:1 (via 10.0.0.1)", b"192.168.1.5:1 (via 10.0.0.1)"),
(b"192.168.1.5:1 (via 8.8.8.8)", b"192.168.1.5:1 (via <REMOVED>)"),
(
b"192.168.1.5:1 (X-Forwarded-For: 8.8.8.8, 10.0.0.2) (via 10.0.0.1)",
b"192.168.1.5:1 (X-Forwarded-For: <REMOVED>, 10.0.0.2) (via 10.0.0.1)",
),
],
)
def test_log_sanitize_remote_label_with_port(self, label, expected):
"""The label carries a port, and an IPv6 one is bracketed. Both have to be matched,
or an external address reaches the log returned by the showlog api-call untouched."""
line = b"2026-05-19 18:35:18,271::WARNING::[interface:689] Refused connection from: %s [Mozilla/5.0]\n" % label
assert api.sanitize_line(line).count(b"%s [Mozilla/5.0]" % expected) == 1
def create_mock_request(
hostname: str = "localhost",
remote_ip: str = "127.0.0.1",
headers: dict | None = None,
query_params: dict | None = None,
):
"""Create a mock Starlette Request object for testing"""
mock_request = Mock(spec=Request)
mock_request.client = Address(remote_ip, 12345)
# What ProxyTrustMiddleware records. A real dict, so the verdict is a bool not a Mock.
mock_request.scope = {
"type": "http",
"client": (remote_ip, 12345),
security.SCOPE_PEER: (remote_ip, 12345),
security.SCOPE_PEER_TRUSTED: False,
}
# Set up headers
request_headers = {"Host": hostname}
if headers:
request_headers.update(headers)
mock_request.headers = Headers(request_headers)
# Set up query params, both on the Request itself and on request.state.params,
# where secured_expose stores the merged GET/POST params (see request_params)
mock_request.query_params = QueryParams(query_params or {})
mock_request.state = State()
mock_request.state.params = mock_request.query_params
return mock_request
def run_get_request_params(method, query_string="", body=b"", content_type=None, merge_query=False):
"""Drive interface.get_request_params with a real Starlette Request"""
headers = [(b"content-type", content_type.encode())] if content_type else []
scope = {
"type": "http",
"method": method,
"query_string": query_string.encode(),
"headers": headers,
}
async def receive():
return {"type": "http.request", "body": body, "more_body": False}
request = Request(scope, receive)
return run_async(interface.get_request_params(request, merge_query=merge_query))
FORM = "application/x-www-form-urlencoded"
class TestGetRequestParams:
"""The /api route must expose GET and POST arguments identically to CherryPy"""
def test_api_duplicate_scalar_key_first_wins(self):
"""A repeated routing/scalar key resolves to its first value, as CherryPy did"""
params = run_get_request_params("GET", "mode=queue&mode=version", merge_query=True)
assert params.get("mode") == "queue"
def test_api_get_and_post_resolve_duplicates_identically(self):
"""GET and a form POST must dispatch a duplicated key the same way"""
get_params = run_get_request_params("GET", "mode=queue&mode=version", merge_query=True)
post_params = run_get_request_params(
"POST", "mode=queue&mode=version", body=b"nzbname=x", content_type=FORM, merge_query=True
)
assert get_params.get("mode") == post_params.get("mode") == "queue"
def test_api_body_wins_over_query(self):
"""On a form POST the body value replaces the query value for the same key"""
params = run_get_request_params(
"POST", "mode=version&apikey=K", body=b"mode=addfile", content_type=FORM, merge_query=True
)
assert params.get("mode") == "addfile"
assert params.get("apikey") == "K"
def test_api_query_only_multi_value_key_preserved(self):
"""A genuinely multi-valued key supplied only in the query keeps every value"""
params = run_get_request_params(
"POST", "keyword=a&keyword=b", body=b"section=misc", content_type=FORM, merge_query=True
)
assert params.getlist("keyword") == ["a", "b"]
def test_api_bodyless_post_uses_query(self):
"""An /api POST without a form body falls back to the query string"""
params = run_get_request_params("POST", "mode=queue&apikey=K", merge_query=True)
assert params.get("mode") == "queue"
assert params.get("apikey") == "K"
def test_page_post_ignores_query_string(self):
"""A page POST (no merge) reads the form body only; the query is not merged in"""
params = run_get_request_params("POST", "smuggled=1", body=b"field=value", content_type=FORM, merge_query=False)
assert params.get("field") == "value"
assert params.get("smuggled") is None
def test_csrf_token_cannot_come_from_the_query_string(self):
"""The CSRF guard reads the token out of these params, so the body-only rule above is
what stops a cross-site form from putting one in the URL instead"""
params = run_get_request_params(
"POST", "csrf_token=smuggled", body=b"field=value", content_type=FORM, merge_query=False
)
assert params.get("csrf_token") is None
class TestOrphanPathTraversal:
"""Orphaned-job handlers must not allow deleting/adding paths outside the download folder"""
@pytest.fixture
def download_dir(self, tmp_path):
"""Point cfg.download_dir at a temporary folder for the duration of a test"""
ddir = tmp_path / "incomplete"
ddir.mkdir()
with mock.patch.object(api.cfg.download_dir, "get_path", return_value=str(ddir)):
yield str(ddir)
# Values that must be silently rejected: absolute paths and '..' traversal escape the
# download folder, an empty value resolves to the download folder itself
traversal_values = [
"/etc/passwd", # absolute path overrides the join base
"../../../etc/passwd", # traversal with ..
"..", # the parent of the download folder
"sub/../../escape", # traversal hidden behind a valid component
"", # resolves to the download folder itself
]
@pytest.mark.parametrize("value", traversal_values)
def test_delete_orphan_does_not_remove_outside(self, download_dir, value):
with mock.patch.object(api, "remove_all") as remove_all_mock:
api._api_delete_orphan(value, {})
remove_all_mock.assert_not_called()
def test_delete_orphan_removes_valid_child(self, download_dir):
with mock.patch.object(api, "remove_all") as remove_all_mock:
api._api_delete_orphan("myjob", {})
remove_all_mock.assert_called_once_with(os.path.join(download_dir, "myjob"), recursive=True)
@pytest.mark.parametrize("value", traversal_values)
def test_add_orphan_does_not_repair_outside(self, download_dir, value):
with mock.patch.object(sabnzbd, "NzbQueue", create=True) as nzbqueue_mock:
api._api_add_orphan(value, {})
nzbqueue_mock.repair_job.assert_not_called()
def test_add_orphan_repairs_valid_child(self, download_dir):
with mock.patch.object(sabnzbd, "NzbQueue", create=True) as nzbqueue_mock:
api._api_add_orphan("myjob", {})
nzbqueue_mock.repair_job.assert_called_once_with(os.path.join(download_dir, "myjob"), None, None)
class TestRetryJobFuturetype:
"""A futuretype job never got past fetching its URL, so there is nothing on disk to
repair: retry_job() has to re-fetch the URL instead of going through repair_job()"""
@pytest.fixture
def history_db(self, tmp_path, monkeypatch):
"""A real history database, also served by the pool that retry_job() borrows from"""
monkeypatch.setattr(db.HistoryDB, "db_path", str(tmp_path / DB_HISTORY_NAME))
monkeypatch.setattr(db.HistoryDB, "startup_done", False)
pool = db.HistoryDBPool(max_connections=1)
monkeypatch.setattr(sabnzbd, "db_pool", pool)
fake_history_db = FakeHistoryDB(db.HistoryDB.db_path)
yield fake_history_db
fake_history_db.close()
pool.close()
@staticmethod
def _add_futuretype_job(history_db) -> tuple[str, tuple]:
"""Add a failed URL-fetch to the history, return its nzo_id and stored settings"""
nzo_id = history_db.add_fake_history_job(
"Ubuntu.Linux.ISO-Usenet", status=Status.FAILED, category="catA", futuretype=True
)
stored_settings = history_db.get_other(nzo_id)
assert stored_settings[0] == "future"
return nzo_id, stored_settings
def test_futuretype_job_url_is_refetched(self, history_db):
nzo_id, (_, url, pp, script, cat) = self._add_futuretype_job(history_db)
with (
mock.patch.object(
sabnzbd.urlgrabber, "add_url", return_value=(AddNzbFileResult.OK, ["new_nzo_id"])
) as add_url_mock,
mock.patch.object(sabnzbd, "NzbQueue", create=True) as nzbqueue_mock,
):
assert api.retry_job(nzo_id) == "new_nzo_id"
# Retried with the settings stored in the history, but without the duplicate check:
# the job that is being retried is still in the history at that point
add_url_mock.assert_called_once_with(url, pp, script, cat, dup_check=False)
# There is no incomplete folder to repair
nzbqueue_mock.repair_job.assert_not_called()
# The old entry is gone, the re-added job will create its own
assert history_db.get_other(nzo_id) == ("", "", "", "", "")
@pytest.mark.parametrize(
"add_url_result",
[
(AddNzbFileResult.NO_FILES_FOUND, []),
(AddNzbFileResult.ERROR, []),
(AddNzbFileResult.RETRY, []),
(AddNzbFileResult.PREQUEUE_REJECTED, []),
],
)
def test_futuretype_job_kept_in_history_when_refetch_fails(self, history_db, add_url_result):
"""Only a job that was actually re-added may leave the history, otherwise the
failure disappears without anything taking its place"""
nzo_id, stored_settings = self._add_futuretype_job(history_db)
with mock.patch.object(sabnzbd.urlgrabber, "add_url", return_value=add_url_result):
assert api.retry_job(nzo_id) is None
assert history_db.get_other(nzo_id) == stored_settings
class TestSecuredExpose:
"""Test the security handling for Starlette interface"""
@cached_property
def main_page(self):
return sabnzbd.interface.MainPage()
def setup_method(self):
"""Set up mocks for SABnzbd components before each test"""
# Instead of mocking every individual component, let's mock the main API functions
# that are used in testing to return simple, predictable responses
# Mock build_queue to return a simple queue response
mock_queue_response = {
"version": "test-version",
"paused": False,
"slots": [],
"noofslots": 0,
"limit": 0,
"start": 0,
"finish": 0,
"cache_art": "0",
"cache_size": "0 B",
"kbpersec": "0.00",
"speed": "0 B/s",
"mbleft": "0.00",
"mb": "0.00",
"sizeleft": "0 B",
"size": "0 B",
"timeleft": "0:00:00",
"eta": "unknown",
}
# Apply patches for main API functions
self.build_queue_patch = patch("sabnzbd.api.build_queue", return_value=mock_queue_response)
# Start all patches
self.build_queue_patch.start()
def teardown_method(self):
"""Clean up mocks after each test"""
self.build_queue_patch.stop()
async def call_api_endpoint(self, request: Request):
"""Call the API endpoint directly"""
return await interface.api(request)
async def call_main_endpoint(self, request: Request):
"""Call the main endpoint directly"""
return await interface.main_index(request)
def api_wrapper(self, **kwargs):
"""Wrapper to test API calls with query parameters"""
request = create_mock_request(query_params=kwargs)
return run_api_handler(request.query_params)
def check_full_access(self, hostname="localhost", remote_ip="127.0.0.1"):
"""Basic test if we have full access to API and interface"""
# Test API access
result = self.api_wrapper(mode="version")
assert sabnzbd.__version__ in result.body.decode()
# Test API with correct key
result = self.api_wrapper(mode="queue", apikey=sabnzbd.cfg.api_key())
assert "queue" in result.body.decode() # Should return queue data
def test_basic(self):
"""Test basic API access functionality"""
self.check_full_access()
def test_api_no_or_wrong_api_key(self):
"""Test API key validation through direct API handler calls"""
# Allowed to access "auth" and "version" without key
result = self.api_wrapper(mode="auth")
assert "apikey" in result.body.decode()
result = self.api_wrapper(mode="version")
assert sabnzbd.__version__ in result.body.decode()
# Other modes should work with correct API key
result = self.api_wrapper(mode="queue", apikey=sabnzbd.cfg.api_key())
assert "queue" in result.body.decode()
def test_api_nzb_key(self):
"""Test NZB key functionality"""
# NZB key should work for addfile (level 1 access)
result = self.api_wrapper(mode="addfile", apikey=sabnzbd.cfg.nzb_key())
assert api._MSG_NO_VALUE in result.body.decode() # No file provided, but key was accepted
def test_check_hostname_basic(self):
"""Test hostname checking functionality"""
# Test the check_hostname_starlette function directly
# Block bad host
bad_request = create_mock_request(hostname="not_me")
assert interface.check_hostname(bad_request) is False
# Block empty hostname
empty_request = create_mock_request(hostname="")
assert interface.check_hostname(empty_request) is False
# Allow valid hostnames/IPs
for test_hostname in (
"100.100.100.100",
"100.100.100.100:8080",
"[2001:db8:3333:4444:5555:6666:7777:8888]",
"[2001:db8:3333:4444:5555:6666:7777:8888]:8080",
"test.local",
"test.local:8080",
"test.local.",
"localhost",
):
good_request = create_mock_request(hostname=test_hostname)
assert interface.check_hostname(good_request) is True
@pytest.mark.config({"username": "foo", "password": "bar"})
def test_check_hostname_with_auth(self):
"""Test hostname checking with authentication enabled"""
# With username/password set, hostname check should always pass
bad_request = create_mock_request(hostname="not_me")
assert interface.check_hostname(bad_request) is True
def test_check_hostname_bare_ipv6_is_refused(self):
"""An IPv6 literal must be bracketed in a Host header (RFC 7230). A bare one is
ambiguous, as there is no telling where the address ends and the port begins,
so it must not be accepted as if the trailing group were a port number."""
for bad_hostname in (
"1234:5678::1:8080",
"bla:bla:1234",
"::ffff:127.0.0.1:8080",
"2001:db8:3333:4444:5555:6666:7777:8888",
):
assert interface.check_hostname(create_mock_request(hostname=bad_hostname)) is False
# The bracketed forms of the same addresses stay allowed
for good_hostname in ("[1234:5678::1]:8080", "[::ffff:127.0.0.1]:8080", "[1234:5678::1]"):
assert interface.check_hostname(create_mock_request(hostname=good_hostname)) is True
@pytest.mark.config({"host_whitelist": "test.com, not_evil"})
def test_check_hostname_whitelist(self):
"""Test hostname whitelist functionality"""
# Whitelisted hostnames should be allowed
request1 = create_mock_request(hostname="test.com")
assert interface.check_hostname(request1) is True
request2 = create_mock_request(hostname="not_evil")
assert interface.check_hostname(request2) is True
# Non-whitelisted hostname should be blocked
request3 = create_mock_request(hostname="evil.com")
assert interface.check_hostname(request3) is False
def test_dual_stack(self):
"""Test IPv6 dual stack functionality"""
request = create_mock_request(remote_ip="::ffff:192.168.0.10")
# Dual stack IPs should be treated as local
assert security.check_access(request, access_type=4) is True
@pytest.mark.config({"local_ranges": "132.10."})
def test_dual_stack_local_ranges(self):
"""Test custom local ranges"""
# IP not in custom local_ranges should be blocked
request1 = create_mock_request(remote_ip="::ffff:192.168.0.10")
assert security.check_access(request1, access_type=5) is False
# IP in custom local_ranges should be allowed
request2 = create_mock_request(remote_ip="::ffff:132.10.0.10")
assert security.check_access(request2, access_type=4) is True
@pytest.mark.config({"inet_exposure": 2})
def test_inet_exposure_basic(self):
"""Test basic inet exposure functionality"""
# Test with external IP (should be blocked for high access levels)
external_request = create_mock_request(remote_ip="11.11.11.11")
# Level 1-2 should be allowed
assert security.check_access(external_request, access_type=1) is True
assert security.check_access(external_request, access_type=2) is True
# Level 3+ should be blocked
assert security.check_access(external_request, access_type=3) is False
assert security.check_access(external_request, access_type=4) is False
@pytest.mark.config({"inet_exposure": 0})
def test_local_access_always_allowed(self):
"""Test that local IPs are always allowed regardless of inet_exposure"""
local_request = create_mock_request(remote_ip="127.0.0.1")
# Even with minimal exposure, local IPs should be allowed
assert security.check_access(local_request, access_type=4) is True
assert security.check_access(local_request, access_type=5) is True
@pytest.mark.parametrize("inet_exposure", [0, 1, 2, 3, 4, 5])
@pytest.mark.parametrize("access_type", [1, 2, 3, 4, 5, 6])
@pytest.mark.parametrize(
"remote_ip,expected_local",
[
("192.168.1.10", True), # Local IP
("127.0.0.1", True), # Loopback IP
("8.8.8.8", False), # External IP
],
)
@pytest.mark.config(lambda params: {"inet_exposure": params["inet_exposure"]})
def test_inet_exposure_levels_comprehensive(self, inet_exposure, access_type, remote_ip, expected_local):
"""Test all inet_exposure levels with different access types and IP types"""
request = create_mock_request(remote_ip=remote_ip)
if expected_local:
# Local and loopback IPs should always be allowed
assert security.check_access(request, access_type) is True
else:
# External IPs should follow inet_exposure rules
expected_allowed = access_type <= inet_exposure
assert security.check_access(request, access_type) is expected_allowed
@pytest.mark.config({"inet_exposure": 2, "verify_xff_header": True})
def test_inet_exposure_with_xff_headers(self):
"""request.client already holds the effective client address, resolved by uvicorn"""
# Local remote IP with external XFF: uvicorn rewrites the client to the
# external XFF address, which should be denied
local_request_external_xff = create_mock_request(
remote_ip=resolve_client(remote_ip="192.168.1.1", xff_header="8.8.8.8").host
)
# Local remote IP with local XFF: effective client is the (local) XFF address
local_request_local_xff = create_mock_request(
remote_ip=resolve_client(remote_ip="192.168.1.1", xff_header="192.168.1.10").host
)
# External remote IP: untrusted peer, XFF is ignored and the client is untouched
external_request = create_mock_request(
remote_ip=resolve_client(remote_ip="8.8.8.8", xff_header="192.168.1.10").host
)
# Local IP with external XFF should be denied
assert security.check_access(local_request_external_xff, access_type=4) is False
# Local IP with local XFF should be allowed
assert security.check_access(local_request_local_xff, access_type=4) is True
# External IP should follow inet_exposure rules (XFF ignored for external IPs)
assert security.check_access(external_request, access_type=1) is True
assert security.check_access(external_request, access_type=2) is True
assert security.check_access(external_request, access_type=3) is False
# Note: The comprehensive parametrized test above covers all these scenarios,
# but this test provides explicit documentation of the API access level meanings
@pytest.mark.config({"inet_exposure": 2})
def test_inet_exposure_api_levels_documentation(self):
"""Document the different API access levels with inet_exposure"""
external_request = create_mock_request(remote_ip="8.8.8.8")
# access_type = 1: NZB upload access
assert security.check_access(external_request, access_type=1) is True
# access_type = 2: Basic API access
assert security.check_access(external_request, access_type=2) is True
# access_type = 3: Full API access (blocked with inet_exposure=2)
assert security.check_access(external_request, access_type=3) is False
# access_type = 4: WebUI access (blocked with inet_exposure=2)
assert security.check_access(external_request, access_type=4) is False
@pytest.mark.config({"inet_exposure": 1})
def test_inet_exposure_ipv6(self):
"""Test IPv6 edge cases for inet_exposure"""
# Test IPv6 addresses
ipv6_external_request = create_mock_request(remote_ip="2001:4860:4860::8888")
ipv6_local_request = create_mock_request(remote_ip="::1")
# Test dual-stack (IPv4-mapped IPv6)
dual_stack_request = create_mock_request(remote_ip="::ffff:192.168.1.10")
# IPv6 loopback should always be allowed
assert security.check_access(ipv6_local_request, access_type=4) is True
# IPv6 external should follow inet_exposure rules
assert security.check_access(ipv6_external_request, access_type=1) is True
assert security.check_access(ipv6_external_request, access_type=2) is False
# Dual-stack should be treated as local
assert security.check_access(dual_stack_request, access_type=4) is True
@pytest.mark.config({"inet_exposure": 1, "local_ranges": ["4.4.4.0/24"]})
def test_inet_exposure_custom_local_ranges(self):
"""Test custom local ranges for inet_exposure"""
# Test with custom local ranges
custom_local_request = create_mock_request(remote_ip="4.4.4.10")
# IP in custom local range should be treated as local
assert security.check_access(custom_local_request, access_type=4) is True
# Note: Boundary conditions are covered by the comprehensive parametrized test
# These tests serve as explicit documentation of the most restrictive/permissive settings
@pytest.mark.config({"inet_exposure": 0})
def test_inet_exposure_most_restrictive(self):
"""Document the most restrictive inet_exposure setting"""
external_request = create_mock_request(remote_ip="1.1.1.1")
# inet_exposure=0: No external access allowed for any access type
assert security.check_access(external_request, access_type=1) is False
@pytest.mark.config({"inet_exposure": 5})
def test_inet_exposure_most_permissive(self):
"""Document the most permissive inet_exposure setting"""
external_request = create_mock_request(remote_ip="1.1.1.1")
# inet_exposure=5: External access allowed for access_type 1-5, but not 6
assert security.check_access(external_request, access_type=5) is True
assert security.check_access(external_request, access_type=6) is False
class TestHistory:
@pytest.mark.usefixtures("run_sabnzbd")
def test_add_active_history_consistency(self):
"""Verify that add_active_history has the same structure as fetch_history"""
history_db = os.path.join(SAB_CACHE_DIR, DEF_ADMIN_DIR, DB_HISTORY_NAME)
with FakeHistoryDB(history_db) as fake_history:
fake_history.add_fake_history_jobs(1)
jobs, _total_items = fake_history.fetch_history()
history_job = jobs[-1]
# Add minimal attributes to create pp-job
nzo = mock.Mock()
nzo.final_name = "test_add_active_history"
nzo.repair, nzo.unpack, nzo.delete = pp_to_opts(choice(list(PP_LOOKUP.keys())))
nzo.download_path = os.path.join(os.path.dirname(db.HistoryDB.db_path), "placeholder_downpath")
nzo.bytes_downloaded = randint(1024, 1024**4)
nzo.unpack_info = {"unpack_info": "placeholder unpack_info line\r\n" * 3}
api.add_active_history([nzo], jobs)
# Make sure the job was added to the list
pp_job = jobs[-1]
assert pp_job["name"] == nzo.final_name
assert pp_job["name"] != history_job["name"]
# Compare the keys, so not the values!
pp_keys = list(pp_job.keys())
pp_keys.sort()
history_keys = list(history_job.keys())
history_keys.sort()
assert pp_keys == history_keys
@pytest.mark.usefixtures("run_sabnzbd")
def test_add_active_history_duplicate(self):
"""Verify that add_active_history does not add duplicate entries"""
history_db = os.path.join(SAB_CACHE_DIR, DEF_ADMIN_DIR, DB_HISTORY_NAME)
with FakeHistoryDB(history_db) as fake_history:
fake_history.add_fake_history_jobs(1)
jobs, total_items = fake_history.fetch_history()
history_job = jobs[-1]
# Add minimal attributes to create pp-job
nzo = mock.Mock()
nzo.nzo_id = history_job["nzo_id"]
nzo.final_name = "test_add_active_history"
nzo.repair, nzo.unpack, nzo.delete = pp_to_opts(choice(list(PP_LOOKUP.keys())))
nzo.download_path = os.path.join(os.path.dirname(db.HistoryDB.db_path), "placeholder_downpath")
nzo.bytes_downloaded = randint(1024, 1024**4)
nzo.unpack_info = {"unpack_info": "placeholder unpack_info line\r\n" * 3}
api.add_active_history([nzo], jobs)
# Make sure the job was not added to the list, a completed entry already exists
assert total_items == len(jobs)
@pytest.fixture
def renderable_header(monkeypatch):
"""Stand in for the runtime singletons build_header reads, so it can be called outside
a running SABnzbd"""
for name in ("Scheduler", "Downloader", "GUIHANDLER", "BPSMeter", "ArticleCache", "NzbQueue"):
monkeypatch.setattr(sabnzbd, name, Mock(), raising=False)
monkeypatch.setattr(sabnzbd.BPSMeter, "quota", 0.0, raising=False)
monkeypatch.setattr(sabnzbd.BPSMeter, "left", 0.0, raising=False)
monkeypatch.setattr(sabnzbd.Downloader, "bandwidth_perc", 0, raising=False)
monkeypatch.setattr(sabnzbd.Downloader, "bandwidth_limit", 0, raising=False)
disk = Mock(free=1.0, size=2.0)
with patch("sabnzbd.api.diskspace", return_value=(disk, disk)):
yield
class TestBuildHeaderCsrfToken:
"""build_header hands the page its CSRF token, but only when it has a request."""
@pytest.mark.config({"username": "", "password": ""})
def test_present_for_a_page_request(self, renderable_header):
request = Mock(spec=Request)
request.state.csrf_token = "a-token"
assert api.build_header(request=request)["csrf_token"] == "a-token"
@pytest.mark.config({"username": "", "password": ""})
def test_absent_without_a_request(self, renderable_header):
# The signature build_status() uses, and the one that reaches an API payload
assert "csrf_token" not in api.build_header(trans_functions=False)
assert "csrf_token" not in api.build_header()
@pytest.mark.config({"username": "", "password": ""})
def test_empty_when_no_middleware_published_one(self, renderable_header):
"""Routes that render without SecurityMiddleware having set it (/login) get an empty
string rather than an AttributeError"""
request = Mock(spec=Request)
request.state = State({})
assert api.build_header(request=request)["csrf_token"] == ""
Reference in new issue View Git Blame Copy Permalink