mirror of
https://github.com/sabnzbd/sabnzbd.git
synced 2026-09-22 02:45:20 -04:00
* Migrate web interface from CherryPy to Uvicorn/Starlette
Squashed rebase of feature/uvicorn (34 commits) onto develop, reconciled
with ~3 months of intervening develop changes.
Replaces the CherryPy webserver and request handling with Uvicorn/Starlette
across the API, web interface, RSS, config pages and related modules.
Reconciliation with develop during the rebase:
- api.py: kept develop's security/behaviour fixes (orphan path-traversal
guard, expanded log redaction incl. host_whitelist and
remote_label_replacement, get_dconfig single-return, get_retryable_jobs,
connections default, translated NNTP test errors) on top of the Starlette
request/response rewrite.
- interface.py: ported the RSS route handlers to develop's DB-backed
RSSRepository API (process_feed, rss_repository / find_job_by_url /
clear_feed / clear_downloaded / flag_downloaded).
- misc.py: kept develop's hachoir-based get_media_duration.
- requirements.txt: dropped the CherryPy stack, adopted develop's newer pins.
Also applied ruff --fix (PEP 604 unions, builtin generics) to align with
develop's lint config.
Verified: ruff check, black --check, and the affected test suites
(9413 passed, 1 skipped) all pass.
* Update starlette/uvicorn versions
* Fix race issues in global rss state
* Fix test race in server shutdown
The uvicorn migration turned /shutdown (and the shutdown API) into fire-and-forget: it spawned shutdown_program() in a background thread and replied immediately, whereas develop ran it synchronously and only replied once halt() had persisted all state. Because the module-scoped test teardown doesn't wait for the process to exit, the next module's clean_cache_dir wiped the shared cache dir (and reused the fixed port) while the previous instance was still saving state and holding the port — producing the three intermittent failures (deleted sabnzbd.log → "File log disabled or not found"; un-persisted [sorters] → KeyError; stale instance → missing wizard .quoteBlock).
* Fix robots and description, add favicon
* Remove remains of http basic auth
* Setup Starlette once configuration is available, fix static file relative cwd and url_base config
* abort_and_show_error when webserver fails to start
* Guard stopping webserver that never started
* Delegate XFF handling to ProxyHeadersMiddleware
* Merged params at request.state.params instead of modifying private apis
* Both shutdown routes share implementation and do not block event loop
* Run sync handlers via run_in_threadpool and facilitate eventual migration to async
* Pool database connections
* Online backup of database due to WAL changes
* Fix exception on None request.client (test clients or unix sockets)
* Fix flakey tests due to process not fully shutting down
* Restore X-Frame-Options behaviour via middleware
* Fix set_config_default with multiple keywords
* Remove broken logging call
* Restore api logging functionality
* Cache-Control: no-store
* Login only via POST
* Remove 401 (basic-auth) and add 404 handling via redirect
* Fix crash when shutdown not an int
* Use BaseRedirectResponse helper
* Remove trailing slashes from wizard routes
* URL helper, absolute URLs everywhere, fixes issues with nested navigation
* Fix scheduler adding multiple daysofweek
* Restore CherryPy api behaviour merging body with query params (body wins)
* Clearer documentation of get_request_params and request_params
* First stage supporting gradual api async
* Fix rss ajax consuming flash
* Restore access log functionality
* Hostname check in middleware
* Request logging in middleware
* Param parsing in middleware
* Security checks in middleware
* secured_expose is now purely route registration
* Lookup api handler once per request
* Fix flakey alert dialogs
* Trigger restart via BackgroundTask
* Restore CherryPy first param wins and get/post consistency
* Remove dead code
* Secure cookies based on protocol the client used
* Fix various issues with port_is_free
1. port_is_free answered the wrong question. It connect-probed ("is something answering?") rather than bind-probed ("can I bind?"). A port could report free and then kill startup at uvicorn's bind().
2. The bind-all remap crossed address families. :: was mapped to 127.0.0.1, probing IPv4 for an IPv6 bind — a regression against portend, which maps :: → ::1.
3. The call sites passed the wrong host. browserhost is a client-reachable address; the thing that has to be bindable is web_host.
4. Errors were swallowed. A bare except OSError hid gaierror, so an unresolvable host reported "free".
5. find_free_port had a port-0 trap. Under a bind-probe, currentport=0 always succeeds and returned 0 — the old failure sentinel. Now guarded, and None instead of 0.
6. Ports 80/443 were misdiagnosed. EACCES was folded into "occupied", producing ten futile probes and a panic claiming another program held the port. PermissionError now propagates to a dedicated panic explaining the actual remedies.
7. The tests were largely tautological. Three tests covering one branch, an IPv6 test with no IPv6 in it, a timeout test that never engaged the timeout, TOCTOU-prone fixed-range probes, no SO_REUSEADDR on the helper listener, and nothing asserting the property that matters — that "free" implies bindable.
8. A portability bug I introduced, then fixed. I'd baked Linux SO_REUSEADDR overlap semantics into four assertions; macOS differs. Now platform-aware, with the IPv6 regression re-covered by checking the socket family directly.
* Claim the bind address for uvicorn on startup, resolves "49" in err handling from cherrypy
* Rename function BaseRedirectResponse to base_redirect_response
* Restore error response on change web directory
* Add missing typings
* Fix return type of retry job for future types
* A better fix for xdist compatibility - test overwrote db_path
* Secure session cookies (rss flash)
* Inline or remove some functions
* Retry job futuretype behaviour
* Sneak a worksteal fix in
* Test and fix retry_job futuretype behaviour
311 lines
14 KiB
Python
311 lines
14 KiB
Python
#!/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_config - Tests of config methods
|
|
"""
|
|
|
|
import io
|
|
import os
|
|
import shutil
|
|
import time
|
|
import zipfile
|
|
|
|
import pytest
|
|
|
|
import sabnzbd
|
|
import sabnzbd.cfg
|
|
import sabnzbd.database
|
|
from sabnzbd import config, filesystem
|
|
from sabnzbd.constants import (
|
|
CONFIG_BACKUP_FILES,
|
|
CONFIG_BACKUP_HTTPS,
|
|
DB_HISTORY_NAME,
|
|
DEF_HTTPS_CERT_FILE,
|
|
DEF_HTTPS_KEY_FILE,
|
|
DEF_INI_FILE,
|
|
)
|
|
from sabnzbd.filesystem import long_path
|
|
from tests.testhelper import SAB_CACHE_DIR, SAB_COMPLETE_DIR, SAB_DATA_DIR
|
|
|
|
DEF_CHAIN_FILE = "server.chain"
|
|
|
|
# Stand-in for the SQLite online backup of the history database. This test only
|
|
# fabricates history1.db as a file, while a real snapshot needs a live database
|
|
# and connection pool, so history_db_snapshot is patched to return these bytes.
|
|
# The snapshot itself is tested in tests/test_database.py.
|
|
FAKE_HISTORY_SNAPSHOT = b"fake history database snapshot"
|
|
|
|
|
|
class TestOptions:
|
|
test_section = "test_section"
|
|
test_keyword = "test_keyword"
|
|
|
|
def test_base_option(self):
|
|
test_option = config.Option(self.test_section, self.test_keyword)
|
|
assert test_option.section == self.test_section
|
|
assert test_option.keyword == self.test_keyword
|
|
assert test_option.section in config.CONFIG.database
|
|
assert test_option.keyword in config.CONFIG.database[test_option.section]
|
|
assert config.CONFIG.database[test_option.section][test_option.keyword] == test_option
|
|
|
|
@pytest.mark.xfail(reason="These tests should be added")
|
|
def test_all(self):
|
|
# Need to add tests for all the relevant options
|
|
raise NotImplementedError
|
|
|
|
def test_non_public(self):
|
|
test_option = config.Option(self.test_section, self.test_keyword, public=True)
|
|
assert test_option.get_dict() == {self.test_keyword: None}
|
|
assert test_option.get_dict(for_public_api=False) == {self.test_keyword: None}
|
|
|
|
test_option = config.Option(self.test_section, self.test_keyword, public=False)
|
|
assert test_option.get_dict() == {self.test_keyword: None}
|
|
assert test_option.get_dict(for_public_api=True) == {}
|
|
|
|
# Password is special when using for_public_api
|
|
test_option = config.OptionPassword(self.test_section, self.test_keyword, default_val="test_password")
|
|
assert test_option.get_dict() == {self.test_keyword: "test_password"}
|
|
assert test_option.get_dict(for_public_api=True) == {self.test_keyword: "**********"}
|
|
|
|
|
|
@pytest.mark.usefixtures("clean_cache_dir")
|
|
class TestConfig:
|
|
@staticmethod
|
|
def create_dummy_zip(filename: str) -> bytes:
|
|
with io.BytesIO() as zip_buffer:
|
|
with zipfile.ZipFile(zip_buffer, "a", zipfile.ZIP_DEFLATED, False) as zip_ref:
|
|
zip_ref.writestr(filename, "foobar")
|
|
return zip_buffer.getvalue()
|
|
|
|
@staticmethod
|
|
def create_and_verify_backup(admin_dir: str, must_haves: list[str]):
|
|
# Create the backup
|
|
config_backup_path = config.create_config_backup()
|
|
assert os.path.exists(config_backup_path)
|
|
assert sabnzbd.__version__ in config_backup_path
|
|
assert time.strftime("%Y.%m.%d_%H") in config_backup_path
|
|
|
|
# Verify the zipfile has the expected content
|
|
with open(config_backup_path, "rb") as fp:
|
|
# Do basic backup validation
|
|
assert config.validate_config_backup(fp.read())
|
|
# Reset the file pointer
|
|
fp.seek(0)
|
|
with zipfile.ZipFile(fp, "r") as zip:
|
|
for basename in must_haves:
|
|
assert zip.getinfo(basename)
|
|
# The history database is stored as an online snapshot, not a raw file copy
|
|
if DB_HISTORY_NAME in must_haves:
|
|
assert zip.read(DB_HISTORY_NAME) == FAKE_HISTORY_SNAPSHOT
|
|
# Make sure there's nothing else in the zip
|
|
assert (zip_len := len(zip.filelist)) == len(must_haves)
|
|
|
|
# Move the current admin dir out of the way
|
|
stowed_admin = os.path.join(SAB_CACHE_DIR, "stowed_admin")
|
|
if os.path.isdir(stowed_admin):
|
|
filesystem.remove_all(stowed_admin)
|
|
assert not os.path.exists(stowed_admin)
|
|
os.rename(admin_dir, stowed_admin)
|
|
assert os.path.exists(stowed_admin)
|
|
assert filesystem.globber(stowed_admin) != []
|
|
assert not os.path.exists(admin_dir)
|
|
filesystem.create_all_dirs(admin_dir)
|
|
assert os.path.exists(admin_dir)
|
|
assert filesystem.globber(admin_dir) == []
|
|
|
|
# Store current test settings, as these may change when restoring a backup
|
|
restore_me = {setting: getattr(sabnzbd.cfg, setting)() for setting in CONFIG_BACKUP_HTTPS.values()}
|
|
|
|
# Restore the backup
|
|
with open(config_backup_path, "rb") as config_backup_fp:
|
|
config.restore_config_backup(config_backup_fp.read())
|
|
|
|
# Check settings results
|
|
restore_changed_settings = False
|
|
for filename, setting in CONFIG_BACKUP_HTTPS.items():
|
|
if filename in must_haves:
|
|
restore_changed_settings = True
|
|
value = getattr(sabnzbd.cfg, setting)()
|
|
if setting != "https_chain":
|
|
# All https settings should point to the default basenames of the restored files...
|
|
assert value == getattr(sabnzbd.cfg, setting).default
|
|
else:
|
|
# ...except the one that doesn't have a default and uses a hardcoded filename instead
|
|
assert value == DEF_CHAIN_FILE
|
|
# Check filename results
|
|
for basename in must_haves:
|
|
# Verify all files in the backup were restored into the admin dir...
|
|
assert os.path.exists(os.path.join(admin_dir, basename))
|
|
# ...and nothing else
|
|
if not restore_changed_settings:
|
|
assert zip_len == len(filesystem.globber(admin_dir))
|
|
else:
|
|
# Account for sabnzbd.ini.bak in case settings were changed as part of the restore
|
|
assert zip_len + 1 == len(filesystem.globber(admin_dir))
|
|
|
|
# Restore the test settings
|
|
for setting, value in restore_me.items():
|
|
getattr(sabnzbd.cfg, setting).set(value)
|
|
sabnzbd.config.save_config(True)
|
|
|
|
# Purge the backup file to prevent collisions
|
|
os.unlink(config_backup_path)
|
|
assert not os.path.exists(config_backup_path)
|
|
|
|
# Call the original admin dir back into active duty
|
|
filesystem.remove_all(admin_dir)
|
|
assert not os.path.exists(admin_dir)
|
|
os.rename(stowed_admin, admin_dir)
|
|
assert os.path.exists(admin_dir)
|
|
assert filesystem.globber(admin_dir) != []
|
|
assert not os.path.exists(stowed_admin)
|
|
|
|
def test_validate_config_backup(self):
|
|
"""Validate basic dummy data"""
|
|
assert not config.validate_config_backup(b"invalid")
|
|
assert not config.validate_config_backup(self.create_dummy_zip("dummyfile"))
|
|
assert config.validate_config_backup(self.create_dummy_zip(DEF_INI_FILE))
|
|
|
|
@pytest.mark.config(
|
|
{
|
|
"admin_dir": os.path.join(SAB_CACHE_DIR, "test_config_backup"),
|
|
"complete_dir": os.path.join(SAB_COMPLETE_DIR, "test_config_backup"),
|
|
}
|
|
)
|
|
def test_config_backup(self, monkeypatch):
|
|
"""Combined tests for the config.{create,validate,restore}_config_backup functions"""
|
|
monkeypatch.setattr(sabnzbd, "CONFIG_BACKUP_HTTPS_OK", [])
|
|
monkeypatch.setattr(sabnzbd.database, "history_db_snapshot", lambda: FAKE_HISTORY_SNAPSHOT)
|
|
|
|
# Prepare the basics
|
|
admin_dir = sabnzbd.cfg.admin_dir.get_path()
|
|
sabnzbd.cfg.set_root_folders2()
|
|
ini_path = os.path.join(admin_dir, DEF_INI_FILE)
|
|
shutil.copyfile(os.path.join(SAB_DATA_DIR, "sabnzbd.basic.ini"), ini_path)
|
|
assert os.path.exists(ini_path)
|
|
config.read_config(ini_path)
|
|
filesystem.create_all_dirs(sabnzbd.cfg.complete_dir())
|
|
assert os.path.exists(sabnzbd.cfg.complete_dir())
|
|
|
|
# Create a backup and verify it has the expected files (ini only, as there are no admin and https config files)
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE])
|
|
|
|
# Add other admin files that qualify for inclusion in backups
|
|
for basename in CONFIG_BACKUP_FILES:
|
|
with open(admin_file := os.path.join(admin_dir, basename), "wb") as fp:
|
|
fp.write(os.urandom(128))
|
|
assert os.path.exists(admin_file)
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE, *CONFIG_BACKUP_FILES])
|
|
|
|
# Add some useless files in the admin_dir
|
|
for basename in ["totals3.sab", "Best.Movie.Ever.1951.240p.avi", "Rating.sab"]:
|
|
with open(useless_file := os.path.join(admin_dir, basename), "wb") as fp:
|
|
fp.write(os.urandom(256))
|
|
assert os.path.exists(useless_file)
|
|
# None of these should appear in the backup
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE, *CONFIG_BACKUP_FILES])
|
|
|
|
# Remove the extra admin files, but keep the useless ones around
|
|
for basename in CONFIG_BACKUP_FILES:
|
|
os.unlink(admin_file := os.path.join(admin_dir, basename))
|
|
assert not os.path.exists(admin_file)
|
|
|
|
# Generate fake HTTPS certificate and key files
|
|
cert_file = os.path.join(admin_dir, DEF_HTTPS_CERT_FILE)
|
|
key_file = os.path.join(admin_dir, DEF_HTTPS_KEY_FILE)
|
|
for filepath in (cert_file, key_file):
|
|
with open(filepath, "wb") as fp:
|
|
fp.write(os.urandom(512))
|
|
assert os.path.exists(cert_file)
|
|
assert os.path.exists(key_file)
|
|
|
|
# Copy cert and key to create a second set of https config files outside the admin dir
|
|
other_cert_file = long_path(os.path.join(SAB_CACHE_DIR, "foobar.mycert"))
|
|
other_key_file = long_path(os.path.join(SAB_CACHE_DIR, "foobar.mykey"))
|
|
shutil.copyfile(cert_file, other_key_file)
|
|
shutil.copyfile(key_file, other_cert_file)
|
|
assert os.path.exists(other_cert_file)
|
|
assert os.path.exists(other_key_file)
|
|
|
|
# Imitate a mainstream https setup (cert and key present, but no chain file)
|
|
sabnzbd.cfg.enable_https.set(True)
|
|
sabnzbd.cfg.https_cert.set(DEF_HTTPS_CERT_FILE)
|
|
sabnzbd.cfg.https_key.set(DEF_HTTPS_KEY_FILE)
|
|
sabnzbd.config.save_config(True)
|
|
assert not sabnzbd.cfg.https_chain()
|
|
assert sabnzbd.CONFIG_BACKUP_HTTPS_OK == []
|
|
|
|
# Results should remain the same, as we didn't fake the results of a startup with https enabled yet
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE])
|
|
|
|
# Results should still remain the same, the startup data lists only bogus files
|
|
sabnzbd.CONFIG_BACKUP_HTTPS_OK = ["/tmp/no.cert", "/lib/fuldstændig_falsk.nøgle", "/etc/存在しないファイル"]
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE])
|
|
|
|
# Now pretend the program started with this config (note: full paths must be used for _OK)
|
|
sabnzbd.CONFIG_BACKUP_HTTPS_OK = [cert_file, key_file]
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE, DEF_HTTPS_CERT_FILE, DEF_HTTPS_KEY_FILE])
|
|
|
|
# Pretend some other files were loaded on startup instead
|
|
sabnzbd.CONFIG_BACKUP_HTTPS_OK = [other_cert_file, other_key_file]
|
|
# Files in the settings no longer match those in _OK; no https config should be in the backup
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE])
|
|
|
|
# Set the full path to a key and cert file outside the admin dir
|
|
sabnzbd.cfg.https_cert.set(other_cert_file)
|
|
sabnzbd.cfg.https_key.set(other_key_file)
|
|
sabnzbd.config.save_config(True)
|
|
# Now the files should be included, albeit under the default names
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE, DEF_HTTPS_CERT_FILE, DEF_HTTPS_KEY_FILE])
|
|
|
|
# Repeat with the "others" removed, so there's nothing (but the ini) left to include in the first place
|
|
for f in (other_cert_file, other_key_file):
|
|
os.unlink(f)
|
|
assert not os.path.exists(other_cert_file)
|
|
assert not os.path.exists(other_key_file)
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE])
|
|
|
|
# Make up a chain file
|
|
chain_file = os.path.join(admin_dir, "ssl-chain.txt")
|
|
shutil.copyfile(cert_file, chain_file)
|
|
assert os.path.exists(chain_file)
|
|
# Update the config and the startup record (mostly)
|
|
sabnzbd.cfg.https_cert.set(cert_file)
|
|
sabnzbd.cfg.https_key.set(key_file)
|
|
sabnzbd.cfg.https_chain.set(chain_file)
|
|
sabnzbd.config.save_config(True)
|
|
sabnzbd.CONFIG_BACKUP_HTTPS_OK = [cert_file, key_file]
|
|
|
|
# There may be a chain file now, but as long as it's not listed in _OK it should be excluded from the backup
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE, DEF_HTTPS_CERT_FILE, DEF_HTTPS_KEY_FILE])
|
|
|
|
# Now it should be included
|
|
sabnzbd.CONFIG_BACKUP_HTTPS_OK.append(chain_file)
|
|
self.create_and_verify_backup(
|
|
admin_dir, [DEF_INI_FILE, DEF_HTTPS_CERT_FILE, DEF_HTTPS_KEY_FILE, DEF_CHAIN_FILE]
|
|
)
|
|
|
|
# Same same but more lonely
|
|
sabnzbd.CONFIG_BACKUP_HTTPS_OK = [chain_file, "/tmp/foobar.exe"]
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE, DEF_CHAIN_FILE])
|
|
|
|
# Disabling https shouldn't make any difference as long as the evidence shows it was active on startup
|
|
sabnzbd.cfg.enable_https.set(False)
|
|
sabnzbd.config.save_config(True)
|
|
self.create_and_verify_backup(admin_dir, [DEF_INI_FILE, DEF_CHAIN_FILE])
|