Files
sabnzbd/tests/test_postproc.py
Safihre b18cdc38f8 Robustly strip HTML tags to prevent XSS
Previous HTML stripping could allow malformed or unclosed tags to pass through, potentially leading to Cross-Site Scripting (XSS) vulnerabilities if completed by the browser.

This change updates the regex to aggressively remove all HTML tags, even incomplete ones. The `knockout-extensions.js` code also now universally escapes HTML, removing a potential bypass. New tests confirm this behavior.
2026-09-10 12:37:01 +02:00

599 lines
25 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_postproc- Tests of various functions in newspack, among which rar_renamer()
"""
import os
import re
import shutil
from unittest import mock
import pytest
import sabnzbd
import sabnzbd.config
from sabnzbd import postproc
from sabnzbd.config import ConfigCat, ConfigSorter
from sabnzbd.filesystem import clip_path, globber_full
from sabnzbd.misc import sort_to_opts
from tests.testhelper import SAB_CACHE_DIR, SAB_DATA_DIR
@pytest.mark.usefixtures("clean_cache_dir")
class TestPostProc:
# Helper function for rar_renamer tests
def _deobfuscate_dir(self, sourcedir, expected_filename_matches):
"""Function to deobfuscate one directory with rar_renamer()"""
# We create a workingdir inside the sourcedir, because the filenames are really changed
workingdir = os.path.join(SAB_CACHE_DIR, "workingdir_test_rar_renamer")
# if workingdir is still there from previous run, remove it:
if os.path.isdir(workingdir):
try:
shutil.rmtree(workingdir)
except PermissionError:
pytest.fail("Could not remove existing workingdir %s for rar_renamer" % workingdir)
# create a fresh copy
try:
shutil.copytree(sourcedir, workingdir)
except Exception:
pytest.fail("Could not create copy of files for rar_renamer")
# And now let the magic happen:
nzo = mock.Mock()
nzo.final_name = "somedownloadname"
nzo.download_path = workingdir
number_renamed_files = postproc.rar_renamer(nzo)
# run check on the resulting files
if expected_filename_matches:
for filename_match in expected_filename_matches:
if len(globber_full(workingdir, filename_match)) != expected_filename_matches[filename_match]:
pytest.fail("Failed filename_match %s in %s" % (filename_match, workingdir))
# Remove workingdir again
try:
shutil.rmtree(workingdir)
except Exception:
pytest.fail("Could not remove existing workingdir %s for rar_renamer" % workingdir)
return number_renamed_files
def test_rar_renamer_obfuscated_single_rar_set(self):
"""Test rar_renamer with obfuscated single rar set"""
sourcedir = os.path.join(SAB_DATA_DIR, "obfuscated_single_rar_set")
# Now define the filematches we want to see, in which amount ("*-*-*-*-*" are the input files):
expected_filename_matches = {"*part007.rar": 1, "*-*-*-*-*": 0}
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 7
def test_rar_renamer_obfuscated_two_rar_sets(self):
"""Test rar_renamer with obfuscated two rar sets"""
sourcedir = os.path.join(SAB_DATA_DIR, "obfuscated_two_rar_sets")
expected_filename_matches = {"*part007.rar": 2, "*part009.rar": 1, "*-*-*-*-*": 0}
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 16
def test_rar_renamer_obfuscated_but_no_rar(self):
"""Test rar_renamer with obfuscated files that are not rar sets"""
sourcedir = os.path.join(SAB_DATA_DIR, "obfuscated_but_no_rar")
expected_filename_matches = {"*.rar": 0, "*-*-*-*-*": 6}
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 0
def test_rar_renamer_single_rar_set_missing_first_rar(self):
"""Test rar_renamer with single rar set missing first rar"""
# One obfuscated rar set, but first rar (.part1.rar) is missing
sourcedir = os.path.join(SAB_DATA_DIR, "obfuscated_single_rar_set_missing_first_rar")
# single rar set (of 6 obfuscated rar files), so we expect renaming
# thus result must 6 rar files, and 0 obfuscated files
expected_filename_matches = {"*.rar": 6, "*-*-*-*-*": 0}
# 6 files should have been renamed
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 6
def test_rar_renamer_double_rar_set_missing_rar(self):
"""Test rar_renamer with two rar sets where some rars are missing"""
# Two obfuscated rar sets, but some rars are missing
sourcedir = os.path.join(SAB_DATA_DIR, "obfuscated_double_rar_set_missing_rar")
# Two sets, missing rar, so we expect no renaming
# thus result should be 0 rar files, and still 8 obfuscated files
expected_filename_matches = {"*.rar": 0, "*-*-*-*-*": 8}
# 0 files should have been renamed
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 0
def test_rar_renamer_fully_encrypted_and_obfuscated(self):
"""Test rar_renamer with fully encrypted and obfuscated rar set"""
# fully encrypted rar-set, and obfuscated rar names
sourcedir = os.path.join(SAB_DATA_DIR, "fully_encrypted_and_obfuscated_rars")
# SABnzbd cannot do anything with this, so we expect no renaming
expected_filename_matches = {"*.rar": 0, "*-*-*-*-*": 6}
# 0 files should have been renamed
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 0
@pytest.mark.parametrize("category", ["testcat", "Default", None])
@pytest.mark.parametrize("has_jobdir", [True, False]) # With or without a job dir
@pytest.mark.parametrize("has_catdir", [True, False]) # Complete directory is defined at category level
@pytest.mark.parametrize("has_active_sorter", [True, False]) # Sorter active for the fake nzo
@pytest.mark.parametrize("sort_string", ["%sn (%r)", "%sn (%r)/file.%ext", ""]) # Identical path result
@pytest.mark.parametrize("marker_file", [None, ".marker"])
@pytest.mark.parametrize("do_folder_rename", [True, False])
@pytest.mark.config(
lambda params: {
"marker_file": params["marker_file"],
"folder_rename": params["do_folder_rename"],
"download_dir": os.path.join(SAB_CACHE_DIR, "incomplete"),
"complete_dir": os.path.join(SAB_CACHE_DIR, "complete"),
}
)
def test_prepare_extraction_path(
self, category, has_jobdir, has_catdir, has_active_sorter, sort_string, marker_file, do_folder_rename
):
# Every parametrization asserts on the exact directory name that comes out, but that
# name depends on what's already on disk. Best effort cleanup.
shutil.rmtree(SAB_CACHE_DIR, ignore_errors=True)
os.makedirs(SAB_CACHE_DIR, exist_ok=True)
# Ensure global CFG_ vars are initialised
sabnzbd.config.read_config(os.devnull)
# Define a sorter and a category (as @pytest.mark.config cannot handle those)
ConfigSorter(
"sorter__test_prepare_extraction_path",
{
"order": 0,
"min_size": 42,
"multipart_label": "",
"sort_string": sort_string,
"sort_cats": [category if category else "no_such_category"],
"sort_type": [
sort_to_opts("all"),
],
"is_active": int(has_active_sorter),
},
)
assert sabnzbd.config.CONFIG.database["sorters"]["sorter__test_prepare_extraction_path"]
if category:
ConfigCat(
category,
{
"order": 0,
"pp": None,
"script": None,
"dir": (
os.path.join(SAB_CACHE_DIR, ("category_dir_for_" + category + ("*" if not has_jobdir else "")))
if has_catdir
else None
),
"newzbin": "",
"priority": 0,
},
)
assert sabnzbd.config.CONFIG.database["categories"][category]
# Mock a minimal nzo, required as function input
fake_nzo = mock.Mock()
fake_nzo.final_name = "FOSS.Rules.S23E06.2160p-SABnzbd"
fake_nzo.cat = category
fake_nzo.nzo_info = {} # Placeholder to prevent a crash in sorting.get_titles()
def _func():
(
tmp_workdir_complete,
workdir_complete,
file_sorter,
not_create_job_dir,
marker_file_result,
) = postproc.prepare_extraction_path(fake_nzo)
tmp_workdir_complete = clip_path(tmp_workdir_complete)
workdir_complete = clip_path(workdir_complete)
# Verify marker file
if marker_file and not not_create_job_dir:
assert marker_file_result
else:
assert not marker_file_result
# Verify sorter
assert file_sorter
if has_active_sorter and category and sort_string:
assert file_sorter.sorter_active
else:
assert not file_sorter.sorter_active
# Verify not_create_job_dir
if category and has_catdir and not has_jobdir and not file_sorter.sorter_active:
assert not_create_job_dir
else:
# Double negatives ftw
assert not not_create_job_dir
# Verify workdir_complete
if not category or not has_catdir:
# Using standard Complete directory as base
assert workdir_complete.startswith(os.path.join(SAB_CACHE_DIR, "complete"))
elif category and has_catdir:
# Based on the category directory
assert workdir_complete.startswith(os.path.join(SAB_CACHE_DIR, "category_dir_for_" + category))
# Check the job directory part (or the lack thereof) as well
if has_active_sorter and category and sort_string:
# Sorter path, with an extra job name work directory inside
assert re.fullmatch(
re.escape(SAB_CACHE_DIR)
+ r".*"
+ re.escape(os.sep)
+ r"Foss Rules \(2160p\)"
+ re.escape(os.sep)
+ fake_nzo.final_name
+ r"(\.\d+)?",
workdir_complete,
)
elif has_jobdir or not (category and has_catdir):
# Standard job name directory
assert re.fullmatch(
re.escape(SAB_CACHE_DIR) + r".*" + re.escape(os.sep) + r"FOSS.Rules.S23E06.2160p-SABnzbd(\.\d+)?",
workdir_complete,
)
else:
# No job directory at all
assert re.fullmatch(
re.escape(SAB_CACHE_DIR) + r".*" + re.escape(os.sep) + r"category_dir_for_([a-zA-Z]+)",
workdir_complete,
)
# Verify tmp_workdir_complete
if do_folder_rename:
if not not_create_job_dir:
assert tmp_workdir_complete != workdir_complete
assert tmp_workdir_complete.replace("_UNPACK_", "") == workdir_complete
else:
assert tmp_workdir_complete == workdir_complete
_func()
@pytest.mark.usefixtures("clean_cache_dir")
class TestGetLastLine:
@pytest.mark.parametrize(
"txt, expected",
[
("", ""),
("plain text", "plain text"),
("first\nsecond\nlast", "last"),
("last\n\n\n", "last"),
("<b>bold</b>", "bold"),
("<a href='x'>link</a>", "link"),
# Unclosed tags must not survive to be completed by later markup
("<img src=x onerror=alert(apiKey)", ""),
("text <script>evil", "text evil"),
("<img src=x onerror=alert(1)\nafter", "after"),
("<div\n<span", ""),
],
)
def test_get_last_line(self, txt, expected):
result = postproc.get_last_line(txt)
assert result == expected
assert "<" not in result
def test_get_last_line_truncates(self):
result = postproc.get_last_line("x" * 200)
assert len(result) == 150
assert result.endswith("...")
class TestCleanupList:
@staticmethod
def _create_file(path):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as f:
f.write(b"data")
assert os.path.exists(path)
return path
@pytest.mark.config({"cleanup_list": ["exe", "nzb"]})
def test_cleanup_list_only_tracked_files(self):
"""Only files of the job may be removed, not other files in the same folder"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_cleanup")
job_files = [self._create_file(os.path.join(base_dir, f)) for f in ("job.mkv", "job.exe", "job.nzb")]
# Files matching the cleanup list, but not part of the job
unrelated_file = self._create_file(os.path.join(base_dir, "MyApplication", "fake.exe"))
# Tracked files that no longer exist shouldn't cause problems
gone_file = os.path.join(base_dir, "gone.exe")
remaining_files = postproc.cleanup_list([*job_files, gone_file], base_dir, skip_nzb=False)
assert remaining_files == [job_files[0], gone_file]
assert os.path.exists(job_files[0])
assert not os.path.exists(job_files[1])
assert not os.path.exists(job_files[2])
assert os.path.exists(unrelated_file)
@pytest.mark.config({"cleanup_list": ["exe", "nzb"]})
def test_cleanup_list_skip_nzb(self):
"""The nzb extension should be ignored when skip_nzb is set"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_cleanup")
job_files = [self._create_file(os.path.join(base_dir, f)) for f in ("job.exe", "job.nzb")]
remaining_files = postproc.cleanup_list(job_files, base_dir, skip_nzb=True)
assert remaining_files == [job_files[1]]
assert not os.path.exists(job_files[0])
assert os.path.exists(job_files[1])
@pytest.mark.config({"cleanup_list": ["nfo"]})
def test_cleanup_list_removes_empty_parent_dirs(self):
"""Directories left empty by the removed files are pruned, but never the base directory"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_cleanup")
job_file = self._create_file(os.path.join(base_dir, "sub", "deeper", "job.nfo"))
# Unrelated empty directory should be left alone
unrelated_dir = os.path.join(base_dir, "empty_dir")
os.makedirs(unrelated_dir, exist_ok=True)
assert postproc.cleanup_list([job_file], base_dir, skip_nzb=False) == []
assert not os.path.exists(os.path.join(base_dir, "sub"))
assert os.path.exists(unrelated_dir)
assert os.path.exists(base_dir)
@pytest.mark.config({"cleanup_list": ["images/*"]})
def test_cleanup_list_path_patterns(self):
"""Path patterns are matched relative to the base directory"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_cleanup")
job_files = [
self._create_file(os.path.join(base_dir, f)) for f in (os.path.join("images", "pic.jpg"), "pic.jpg")
]
remaining_files = postproc.cleanup_list(job_files, base_dir, skip_nzb=False)
assert remaining_files == [job_files[1]]
assert not os.path.exists(job_files[0])
assert os.path.exists(job_files[1])
@pytest.mark.config({"cleanup_list": []})
def test_cleanup_list_disabled(self):
"""Without a cleanup list, nothing is removed"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_cleanup")
job_files = [self._create_file(os.path.join(base_dir, "job.exe"))]
assert postproc.cleanup_list(job_files, base_dir, skip_nzb=False) == job_files
assert os.path.exists(job_files[0])
@pytest.mark.usefixtures("clean_cache_dir")
class TestRemoveUnwantedFiles:
@staticmethod
def _create_file(path):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as f:
f.write(b"data")
assert os.path.exists(path)
return path
@staticmethod
def _fake_nzo(unwanted_ext=0):
fake_nzo = mock.Mock()
fake_nzo.final_name = "TestDownload"
fake_nzo.unwanted_ext = unwanted_ext
return fake_nzo
@pytest.mark.config({"unwanted_extensions": ["exe"], "action_on_unwanted_extensions": 2})
def test_remove_unwanted_files_only_tracked_files(self):
"""Only files of the job may be removed, not other files in the same folder"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
job_files = [self._create_file(os.path.join(base_dir, f)) for f in ("job.mkv", "job.exe")]
# File matching the unwanted extensions, but not part of the job
unrelated_file = self._create_file(os.path.join(base_dir, "MyApplication", "fake.exe"))
# Tracked files that no longer exist shouldn't cause problems
gone_file = os.path.join(base_dir, "gone.exe")
fake_nzo = self._fake_nzo()
remaining_files = postproc.remove_unwanted_files(fake_nzo, [*job_files, gone_file], base_dir)
assert remaining_files == [job_files[0], gone_file]
assert os.path.exists(job_files[0])
assert not os.path.exists(job_files[1])
assert os.path.exists(unrelated_file)
fake_nzo.set_unpack_info.assert_called_once()
@pytest.mark.config({"unwanted_extensions": ["exe"], "action_on_unwanted_extensions": 1})
def test_remove_unwanted_files_removes_empty_parent_dirs(self):
"""Directories left empty by the removed files are pruned, but never the base directory"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
job_file = self._create_file(os.path.join(base_dir, "sub", "deeper", "job.exe"))
assert postproc.remove_unwanted_files(self._fake_nzo(), [job_file], base_dir) == []
assert not os.path.exists(os.path.join(base_dir, "sub"))
assert os.path.exists(base_dir)
@pytest.mark.config(
{"unwanted_extensions": ["mkv", "srt"], "unwanted_extensions_mode": 1, "action_on_unwanted_extensions": 2}
)
def test_remove_unwanted_files_whitelist_mode(self):
"""In whitelist mode anything not listed is removed, except files without extension"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
job_files = [self._create_file(os.path.join(base_dir, f)) for f in ("job.mkv", "job.exe", "no_extension")]
remaining_files = postproc.remove_unwanted_files(self._fake_nzo(), job_files, base_dir)
assert remaining_files == [job_files[0], job_files[2]]
assert os.path.exists(job_files[0])
assert not os.path.exists(job_files[1])
assert os.path.exists(job_files[2])
@pytest.mark.config({"unwanted_extensions": ["exe"], "action_on_unwanted_extensions": 0})
def test_remove_unwanted_files_no_action_configured(self):
"""Without an action configured, nothing is removed"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
job_files = [self._create_file(os.path.join(base_dir, "job.exe"))]
fake_nzo = self._fake_nzo()
assert postproc.remove_unwanted_files(fake_nzo, job_files, base_dir) == job_files
assert os.path.exists(job_files[0])
fake_nzo.set_unpack_info.assert_not_called()
@pytest.mark.config({"unwanted_extensions": [], "action_on_unwanted_extensions": 2})
def test_remove_unwanted_files_no_extensions_configured(self):
"""Without unwanted extensions configured, nothing is removed"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
job_files = [self._create_file(os.path.join(base_dir, "job.exe"))]
assert postproc.remove_unwanted_files(self._fake_nzo(), job_files, base_dir) == job_files
assert os.path.exists(job_files[0])
@pytest.mark.config({"unwanted_extensions": ["exe"], "action_on_unwanted_extensions": 2})
def test_remove_unwanted_files_user_override(self):
"""Skip the check after the user resumed a job paused for an unwanted extension"""
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
job_files = [self._create_file(os.path.join(base_dir, "job.exe"))]
assert postproc.remove_unwanted_files(self._fake_nzo(unwanted_ext=2), job_files, base_dir) == job_files
assert os.path.exists(job_files[0])
class TestNzbOnlyDownload:
@mock.patch("sabnzbd.postproc.process_single_nzb")
@mock.patch("sabnzbd.postproc.listdir_full")
def test_process_nzb_only_download_single_nzb(self, mock_listdir, mock_process_single_nzb):
"""Test process_nzb_only_download with a single NZB file"""
# Setup mock NZO
fake_nzo = mock.Mock()
fake_nzo.final_name = "TestDownload"
fake_nzo.pp = 3
fake_nzo.script = "test_script.py"
fake_nzo.cat = "movies"
fake_nzo.url = "http://example.com/test.nzb"
fake_nzo.priority = 0
# Mock single NZB file
workdir = os.path.join(SAB_CACHE_DIR, "test_workdir")
nzb_file = os.path.join(workdir, "test.nzb")
mock_listdir.return_value = [nzb_file]
# Call the function
result = postproc.process_nzb_only_download(workdir, fake_nzo)
# Verify result
assert result == [nzb_file]
# Verify process_single_nzb was called with correct arguments
mock_process_single_nzb.assert_called_once_with(
"test.nzb",
nzb_file,
pp=3,
script="test_script.py",
cat="movies",
url="http://example.com/test.nzb",
priority=0,
nzbname="TestDownload",
dup_check=False,
)
@mock.patch("sabnzbd.postproc.process_single_nzb")
@mock.patch("sabnzbd.postproc.listdir_full")
def test_process_nzb_only_download_multiple_nzbs(self, mock_listdir, mock_process_single_nzb):
"""Test process_nzb_only_download with multiple NZB files"""
# Setup mock NZO
fake_nzo = mock.Mock()
fake_nzo.final_name = "TestDownload"
fake_nzo.pp = 2
fake_nzo.script = None
fake_nzo.cat = "tv"
fake_nzo.url = "http://example.com/test.nzb"
fake_nzo.priority = 1
# Mock multiple NZB files
workdir = os.path.join(SAB_CACHE_DIR, "test_workdir")
first_nzb = os.path.join(workdir, "first.nzb")
second_nzb = os.path.join(workdir, "second.nzb")
mock_listdir.return_value = [first_nzb, second_nzb]
# Call the function
result = postproc.process_nzb_only_download(workdir, fake_nzo)
# Verify result
assert result == [first_nzb, second_nzb]
# Verify process_single_nzb was called twice with correct arguments
assert mock_process_single_nzb.call_count == 2
mock_process_single_nzb.assert_any_call(
"first.nzb",
first_nzb,
pp=2,
script=None,
cat="tv",
url="http://example.com/test.nzb",
priority=1,
nzbname="TestDownload - first.nzb",
dup_check=False,
)
mock_process_single_nzb.assert_any_call(
"second.nzb",
second_nzb,
pp=2,
script=None,
cat="tv",
url="http://example.com/test.nzb",
priority=1,
nzbname="TestDownload - second.nzb",
dup_check=False,
)
@mock.patch("sabnzbd.postproc.process_single_nzb")
@mock.patch("sabnzbd.postproc.listdir_full")
def test_process_nzb_only_download_mixed_files(self, mock_listdir, mock_process_single_nzb):
"""Test process_nzb_only_download with mixed file types returns None"""
# Setup mock NZO
fake_nzo = mock.Mock()
fake_nzo.final_name = "TestDownload"
# Mock mixed files (NZB and non-NZB)
workdir = os.path.join(SAB_CACHE_DIR, "test_workdir")
mock_listdir.return_value = [
os.path.join(workdir, "test.nzb"),
os.path.join(workdir, "readme.txt"),
]
# Call the function
result = postproc.process_nzb_only_download(workdir, fake_nzo)
# Verify result is None (not NZB-only)
assert result is None
# Verify process_single_nzb was NOT called
mock_process_single_nzb.assert_not_called()
@mock.patch("sabnzbd.postproc.process_single_nzb")
@mock.patch("sabnzbd.postproc.listdir_full")
def test_process_nzb_only_download_empty_directory(self, mock_listdir, mock_process_single_nzb):
"""Test process_nzb_only_download with empty directory returns None"""
# Setup mock NZO
fake_nzo = mock.Mock()
fake_nzo.final_name = "TestDownload"
# Mock empty directory
workdir = os.path.join(SAB_CACHE_DIR, "test_workdir")
mock_listdir.return_value = []
# Call the function
result = postproc.process_nzb_only_download(workdir, fake_nzo)
# Verify result is None (no files)
assert result is None
# Verify process_single_nzb was NOT called
mock_process_single_nzb.assert_not_called()