mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-09-24 22:05:20 +01:00
Patch: Certificate validation setting + Misc fixes (#642)
- Add certificate validation setting - Fix some OIDC providers not linking emails to local users - Reintroduce sort by peers option for prowlarr results - Fix "All languages" search query reverting to default language - Fix download/request dismissal with multiple admin users - Fix download / request behavior on details modal
This commit is contained in:
Alex
@@ -309,7 +309,7 @@ class TestSecuritySettings:
|
||||
action = next((f for f in fields if f.key == "open_users_tab"), None)
|
||||
assert action is not None
|
||||
assert action.label == "Go to Users"
|
||||
assert action.show_when == {"field": "AUTH_METHOD", "value": "builtin"}
|
||||
assert action.show_when == {"field": "AUTH_METHOD", "value": ["builtin", "oidc"]}
|
||||
|
||||
|
||||
class TestSecurityOnSave:
|
||||
|
||||
@@ -279,6 +279,40 @@ class TestActivityRoutes:
|
||||
assert "expired-task-1" in response.json["status"]["complete"]
|
||||
assert response.json["status"]["complete"]["expired-task-1"]["id"] == "expired-task-1"
|
||||
|
||||
def test_admin_snapshot_backfills_terminal_downloads_across_users(self, main_module, client):
|
||||
admin = _create_user(main_module, prefix="admin", role="admin")
|
||||
request_owner = _create_user(main_module, prefix="reader")
|
||||
_set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True)
|
||||
|
||||
main_module.activity_service.record_terminal_snapshot(
|
||||
user_id=request_owner["id"],
|
||||
item_type="download",
|
||||
item_key="download:cross-user-expired-task",
|
||||
origin="requested",
|
||||
final_status="complete",
|
||||
source_id="cross-user-expired-task",
|
||||
snapshot={
|
||||
"kind": "download",
|
||||
"download": {
|
||||
"id": "cross-user-expired-task",
|
||||
"title": "Cross User Task",
|
||||
"author": "Another User",
|
||||
"added_time": 123,
|
||||
"status_message": "Finished",
|
||||
"source": "direct_download",
|
||||
"user_id": request_owner["id"],
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
|
||||
with patch.object(main_module.backend, "queue_status", return_value=_sample_status_payload()):
|
||||
response = client.get("/api/activity/snapshot")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "cross-user-expired-task" in response.json["status"]["complete"]
|
||||
assert response.json["status"]["complete"]["cross-user-expired-task"]["id"] == "cross-user-expired-task"
|
||||
|
||||
def test_snapshot_clears_stale_download_dismissal_when_same_task_is_active(self, main_module, client):
|
||||
user = _create_user(main_module, prefix="reader")
|
||||
_set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False)
|
||||
@@ -332,6 +366,29 @@ class TestActivityRoutes:
|
||||
assert snapshot_two.status_code == 200
|
||||
assert {"item_type": "download", "item_key": "download:shared-task"} not in snapshot_two.json["dismissed"]
|
||||
|
||||
def test_admin_request_dismissal_is_shared_across_admin_users(self, main_module, client):
|
||||
admin_one = _create_user(main_module, prefix="admin-one", role="admin")
|
||||
admin_two = _create_user(main_module, prefix="admin-two", role="admin")
|
||||
|
||||
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
|
||||
_set_session(client, user_id=admin_one["username"], db_user_id=admin_one["id"], is_admin=True)
|
||||
dismiss_response = client.post(
|
||||
"/api/activity/dismiss",
|
||||
json={"item_type": "request", "item_key": "request:999999"},
|
||||
)
|
||||
assert dismiss_response.status_code == 200
|
||||
|
||||
_set_session(client, user_id=admin_two["username"], db_user_id=admin_two["id"], is_admin=True)
|
||||
with patch.object(main_module.backend, "queue_status", return_value=_sample_status_payload()):
|
||||
snapshot_response = client.get("/api/activity/snapshot")
|
||||
history_response = client.get("/api/activity/history?limit=50&offset=0")
|
||||
|
||||
assert snapshot_response.status_code == 200
|
||||
assert {"item_type": "request", "item_key": "request:999999"} in snapshot_response.json["dismissed"]
|
||||
|
||||
assert history_response.status_code == 200
|
||||
assert any(row["item_key"] == "request:999999" for row in history_response.json)
|
||||
|
||||
def test_history_paging_is_stable_and_non_overlapping(self, main_module, client):
|
||||
user = _create_user(main_module, prefix="history-user")
|
||||
_set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False)
|
||||
|
||||
@@ -235,7 +235,10 @@ class TestActivityService:
|
||||
item_key="download:task-2",
|
||||
)
|
||||
|
||||
rows = activity_service.get_undismissed_terminal_downloads(user["id"])
|
||||
rows = activity_service.get_undismissed_terminal_downloads(
|
||||
user["id"],
|
||||
owner_user_id=user["id"],
|
||||
)
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["item_key"] == "download:task-1"
|
||||
assert rows[0]["final_status"] == "complete"
|
||||
@@ -243,3 +246,50 @@ class TestActivityService:
|
||||
"kind": "download",
|
||||
"download": {"id": "task-1", "status_message": "done"},
|
||||
}
|
||||
|
||||
def test_get_undismissed_terminal_downloads_can_span_owners_for_admin_viewer(
|
||||
self,
|
||||
user_db,
|
||||
activity_service,
|
||||
):
|
||||
viewer = user_db.create_user(username="admin-viewer", role="admin")
|
||||
owner_one = user_db.create_user(username="owner-one")
|
||||
owner_two = user_db.create_user(username="owner-two")
|
||||
|
||||
activity_service.record_terminal_snapshot(
|
||||
user_id=owner_one["id"],
|
||||
item_type="download",
|
||||
item_key="download:owner-one-task",
|
||||
origin="direct",
|
||||
final_status="complete",
|
||||
source_id="owner-one-task",
|
||||
terminal_at="2026-01-01T10:00:00+00:00",
|
||||
snapshot={"kind": "download", "download": {"id": "owner-one-task"}},
|
||||
)
|
||||
activity_service.record_terminal_snapshot(
|
||||
user_id=owner_two["id"],
|
||||
item_type="download",
|
||||
item_key="download:owner-two-task",
|
||||
origin="direct",
|
||||
final_status="complete",
|
||||
source_id="owner-two-task",
|
||||
terminal_at="2026-01-01T11:00:00+00:00",
|
||||
snapshot={"kind": "download", "download": {"id": "owner-two-task"}},
|
||||
)
|
||||
activity_service.dismiss_item(
|
||||
user_id=viewer["id"],
|
||||
item_type="download",
|
||||
item_key="download:owner-two-task",
|
||||
)
|
||||
|
||||
all_owner_rows = activity_service.get_undismissed_terminal_downloads(
|
||||
viewer["id"],
|
||||
owner_user_id=None,
|
||||
)
|
||||
assert [row["item_key"] for row in all_owner_rows] == ["download:owner-one-task"]
|
||||
|
||||
owner_one_rows = activity_service.get_undismissed_terminal_downloads(
|
||||
viewer["id"],
|
||||
owner_user_id=owner_one["id"],
|
||||
)
|
||||
assert [row["item_key"] for row in owner_one_rows] == ["download:owner-one-task"]
|
||||
|
||||
@@ -259,3 +259,51 @@ class TestProvisionOIDCUser:
|
||||
assert user["username"] != "john" # Should have a suffix
|
||||
assert user["oidc_subject"] == "sub-456"
|
||||
assert user["auth_source"] == "oidc"
|
||||
|
||||
def test_provision_links_to_existing_user_by_email(self, user_db):
|
||||
"""When allow_email_link=True and emails match, link to existing local user."""
|
||||
from shelfmark.core.oidc_auth import provision_oidc_user
|
||||
user_db.create_user(
|
||||
username="localuser",
|
||||
email="shared@example.com",
|
||||
password_hash="hash",
|
||||
)
|
||||
|
||||
user_info = {
|
||||
"oidc_subject": "oidc-sub-789",
|
||||
"username": "oidcuser",
|
||||
"email": "shared@example.com",
|
||||
"display_name": "OIDC User",
|
||||
}
|
||||
user = provision_oidc_user(
|
||||
user_db, user_info, is_admin=False, allow_email_link=True,
|
||||
)
|
||||
assert user["username"] == "localuser"
|
||||
assert user["oidc_subject"] == "oidc-sub-789"
|
||||
assert user["auth_source"] == "oidc"
|
||||
assert user["email"] == "shared@example.com"
|
||||
|
||||
def test_provision_does_not_link_by_email_when_disabled(self, user_db):
|
||||
"""When allow_email_link=False (default), don't link by email."""
|
||||
from shelfmark.core.oidc_auth import provision_oidc_user
|
||||
user_db.create_user(
|
||||
username="localuser",
|
||||
email="shared@example.com",
|
||||
password_hash="hash",
|
||||
)
|
||||
|
||||
user_info = {
|
||||
"oidc_subject": "oidc-sub-no-link",
|
||||
"username": "oidcuser",
|
||||
"email": "shared@example.com",
|
||||
"display_name": "OIDC User",
|
||||
}
|
||||
user = provision_oidc_user(
|
||||
user_db, user_info, is_admin=False, allow_email_link=False,
|
||||
)
|
||||
# Should create a new user, not link to existing
|
||||
assert user["username"] == "oidcuser"
|
||||
assert user["oidc_subject"] == "oidc-sub-no-link"
|
||||
|
||||
original = user_db.get_user(username="localuser")
|
||||
assert original["oidc_subject"] is None
|
||||
|
||||
@@ -253,13 +253,14 @@ class TestOIDCCallbackEndpoint:
|
||||
assert "issuer validation failed" in error
|
||||
|
||||
@patch("shelfmark.core.oidc_routes._get_oidc_client")
|
||||
def test_callback_redirects_when_auto_provision_disabled(self, mock_get_client, client):
|
||||
def test_callback_redirects_when_auto_provision_disabled_and_no_email_match(
|
||||
self, mock_get_client, client
|
||||
):
|
||||
config = {**MOCK_OIDC_CONFIG, "OIDC_AUTO_PROVISION": False}
|
||||
fake_client = Mock()
|
||||
fake_client.authorize_access_token.return_value = {
|
||||
"userinfo": {
|
||||
"sub": "unknown-user",
|
||||
"email": "unknown@example.com",
|
||||
"preferred_username": "unknown",
|
||||
"groups": [],
|
||||
}
|
||||
@@ -272,7 +273,7 @@ class TestOIDCCallbackEndpoint:
|
||||
assert "Account not found" in error
|
||||
|
||||
@patch("shelfmark.core.oidc_routes._get_oidc_client")
|
||||
def test_callback_allows_pre_created_user_by_verified_email_when_no_provision(
|
||||
def test_callback_links_pre_created_user_by_email_when_no_provision(
|
||||
self, mock_get_client, client, user_db
|
||||
):
|
||||
config = {**MOCK_OIDC_CONFIG, "OIDC_AUTO_PROVISION": False}
|
||||
@@ -283,7 +284,6 @@ class TestOIDCCallbackEndpoint:
|
||||
"userinfo": {
|
||||
"sub": "oidc-alice-sub",
|
||||
"email": "alice@example.com",
|
||||
"email_verified": True,
|
||||
"preferred_username": "alice_oidc",
|
||||
"groups": [],
|
||||
}
|
||||
@@ -298,18 +298,16 @@ class TestOIDCCallbackEndpoint:
|
||||
assert sess.get("db_user_id") is not None
|
||||
|
||||
@patch("shelfmark.core.oidc_routes._get_oidc_client")
|
||||
def test_callback_does_not_link_unverified_email_when_no_provision(
|
||||
def test_callback_does_not_link_when_no_email_and_no_provision(
|
||||
self, mock_get_client, client, user_db
|
||||
):
|
||||
config = {**MOCK_OIDC_CONFIG, "OIDC_AUTO_PROVISION": False}
|
||||
user = user_db.create_user(username="bob", email="bob@example.com", password_hash="hash")
|
||||
user_db.create_user(username="bob", email="bob@example.com", password_hash="hash")
|
||||
|
||||
fake_client = Mock()
|
||||
fake_client.authorize_access_token.return_value = {
|
||||
"userinfo": {
|
||||
"sub": "oidc-bob-sub",
|
||||
"email": "bob@example.com",
|
||||
"email_verified": False,
|
||||
"preferred_username": "bob_oidc",
|
||||
"groups": [],
|
||||
}
|
||||
@@ -321,7 +319,7 @@ class TestOIDCCallbackEndpoint:
|
||||
assert error is not None
|
||||
assert "Account not found" in error
|
||||
|
||||
updated_user = user_db.get_user(user_id=user["id"])
|
||||
updated_user = user_db.get_user(username="bob")
|
||||
assert updated_user["oidc_subject"] is None
|
||||
|
||||
@patch("shelfmark.core.oidc_routes._get_oidc_client")
|
||||
@@ -359,3 +357,84 @@ class TestOIDCCallbackEndpoint:
|
||||
error = _get_oidc_error(resp)
|
||||
assert error is not None
|
||||
assert "Authentication failed" in error
|
||||
|
||||
@patch("shelfmark.core.oidc_routes._get_oidc_client")
|
||||
def test_callback_links_to_existing_user_by_email(
|
||||
self, mock_get_client, client, user_db
|
||||
):
|
||||
"""OIDC login with matching email should link to existing local user."""
|
||||
user_db.create_user(username="localuser", email="shared@example.com", password_hash="hash")
|
||||
|
||||
fake_client = Mock()
|
||||
fake_client.authorize_access_token.return_value = {
|
||||
"userinfo": {
|
||||
"sub": "oidc-new-sub",
|
||||
"email": "shared@example.com",
|
||||
"preferred_username": "oidcuser",
|
||||
"groups": [],
|
||||
}
|
||||
}
|
||||
mock_get_client.return_value = (fake_client, MOCK_OIDC_CONFIG)
|
||||
|
||||
resp = client.get("/api/auth/oidc/callback?code=abc123&state=test-state")
|
||||
assert resp.status_code == 302
|
||||
|
||||
with client.session_transaction() as sess:
|
||||
assert sess["user_id"] == "localuser"
|
||||
|
||||
linked = user_db.get_user(username="localuser")
|
||||
assert linked["oidc_subject"] == "oidc-new-sub"
|
||||
assert linked["auth_source"] == "oidc"
|
||||
|
||||
@patch("shelfmark.core.oidc_routes._get_oidc_client")
|
||||
def test_callback_creates_new_user_when_no_email_match(
|
||||
self, mock_get_client, client, user_db
|
||||
):
|
||||
"""OIDC login without matching email creates a new user."""
|
||||
user_db.create_user(username="existing", email="other@example.com", password_hash="hash")
|
||||
|
||||
fake_client = Mock()
|
||||
fake_client.authorize_access_token.return_value = {
|
||||
"userinfo": {
|
||||
"sub": "oidc-nomatch",
|
||||
"email": "different@example.com",
|
||||
"preferred_username": "newuser",
|
||||
"groups": [],
|
||||
}
|
||||
}
|
||||
mock_get_client.return_value = (fake_client, MOCK_OIDC_CONFIG)
|
||||
|
||||
resp = client.get("/api/auth/oidc/callback?code=abc123&state=test-state")
|
||||
assert resp.status_code == 302
|
||||
|
||||
with client.session_transaction() as sess:
|
||||
assert sess["user_id"] == "newuser"
|
||||
|
||||
original = user_db.get_user(username="existing")
|
||||
assert original["oidc_subject"] is None
|
||||
|
||||
@patch("shelfmark.core.oidc_routes._get_oidc_client")
|
||||
def test_callback_no_email_link_when_oidc_has_no_email(
|
||||
self, mock_get_client, client, user_db
|
||||
):
|
||||
"""OIDC login without email in claims should not attempt email linking."""
|
||||
user_db.create_user(username="existing", email="existing@example.com", password_hash="hash")
|
||||
|
||||
fake_client = Mock()
|
||||
fake_client.authorize_access_token.return_value = {
|
||||
"userinfo": {
|
||||
"sub": "oidc-noemail",
|
||||
"preferred_username": "noemailuser",
|
||||
"groups": [],
|
||||
}
|
||||
}
|
||||
mock_get_client.return_value = (fake_client, MOCK_OIDC_CONFIG)
|
||||
|
||||
resp = client.get("/api/auth/oidc/callback?code=abc123&state=test-state")
|
||||
assert resp.status_code == 302
|
||||
|
||||
with client.session_transaction() as sess:
|
||||
assert sess["user_id"] == "noemailuser"
|
||||
|
||||
original = user_db.get_user(username="existing")
|
||||
assert original["oidc_subject"] is None
|
||||
|
||||
@@ -2,10 +2,11 @@ import requests
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
def __init__(self, status_code: int, *, headers: dict | None = None, text: str = "") -> None:
|
||||
def __init__(self, status_code: int, *, headers: dict | None = None, text: str = "", url: str = "") -> None:
|
||||
self.status_code = status_code
|
||||
self.headers = headers or {}
|
||||
self.text = text
|
||||
self.url = url
|
||||
|
||||
@property
|
||||
def is_redirect(self) -> bool: # requests.Response compatibility
|
||||
@@ -53,9 +54,9 @@ def test_html_get_page_aa_cross_host_redirect_rotates_mirror(monkeypatch):
|
||||
def fake_get(url: str, **kwargs):
|
||||
calls.append({"url": url, "allow_redirects": kwargs.get("allow_redirects")})
|
||||
if url.startswith("https://annas-archive.li/"):
|
||||
return _FakeResponse(302, headers={"Location": "https://annas-archive.pm/search?q=test"})
|
||||
return _FakeResponse(302, headers={"Location": "https://annas-archive.pm/search?q=test"}, url=url)
|
||||
if url.startswith("https://annas-archive.gl/"):
|
||||
return _FakeResponse(200, text="OK")
|
||||
return _FakeResponse(200, text="OK", url=url)
|
||||
raise AssertionError(f"Unexpected URL: {url}")
|
||||
|
||||
monkeypatch.setattr(http.requests, "get", fake_get)
|
||||
@@ -88,9 +89,9 @@ def test_html_get_page_aa_same_host_redirect_is_followed(monkeypatch):
|
||||
def fake_get(url: str, **kwargs):
|
||||
calls.append({"url": url, "allow_redirects": kwargs.get("allow_redirects")})
|
||||
if url == "https://annas-archive.li/search?q=test":
|
||||
return _FakeResponse(302, headers={"Location": "/search?q=test&page=1"})
|
||||
return _FakeResponse(302, headers={"Location": "/search?q=test&page=1"}, url=url)
|
||||
if url == "https://annas-archive.li/search?q=test&page=1":
|
||||
return _FakeResponse(200, text="OK2")
|
||||
return _FakeResponse(200, text="OK2", url=url)
|
||||
raise AssertionError(f"Unexpected URL: {url}")
|
||||
|
||||
monkeypatch.setattr(http.requests, "get", fake_get)
|
||||
@@ -125,7 +126,7 @@ def test_html_get_page_locked_aa_does_not_fail_over_on_cross_host_redirect(monke
|
||||
def fake_get(url: str, **kwargs):
|
||||
calls.append(url)
|
||||
if url.startswith("https://annas-archive.li/"):
|
||||
return _FakeResponse(302, headers={"Location": "https://annas-archive.pm/search?q=test"})
|
||||
return _FakeResponse(302, headers={"Location": "https://annas-archive.pm/search?q=test"}, url=url)
|
||||
raise AssertionError(f"Unexpected URL: {url}")
|
||||
|
||||
monkeypatch.setattr(http.requests, "get", fake_get)
|
||||
|
||||
@@ -0,0 +1,274 @@
|
||||
"""Tests for certificate validation / SSL verify utilities."""
|
||||
|
||||
import warnings
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_ssl_verify()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetSslVerify:
|
||||
"""Tests for get_ssl_verify() return values across all modes."""
|
||||
|
||||
def test_enabled_returns_true(self, monkeypatch):
|
||||
import shelfmark.download.network as network
|
||||
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "enabled" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
assert network.get_ssl_verify("https://example.com") is True
|
||||
|
||||
def test_enabled_returns_true_for_local_url(self, monkeypatch):
|
||||
import shelfmark.download.network as network
|
||||
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "enabled" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
assert network.get_ssl_verify("https://localhost:8080") is True
|
||||
|
||||
def test_disabled_returns_false_for_public_url(self, monkeypatch):
|
||||
import shelfmark.download.network as network
|
||||
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "disabled" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
assert network.get_ssl_verify("https://example.com") is False
|
||||
|
||||
def test_disabled_returns_false_for_local_url(self, monkeypatch):
|
||||
import shelfmark.download.network as network
|
||||
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "disabled" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
assert network.get_ssl_verify("https://192.168.1.1:9091") is False
|
||||
|
||||
def test_disabled_returns_false_with_no_url(self, monkeypatch):
|
||||
import shelfmark.download.network as network
|
||||
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "disabled" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
assert network.get_ssl_verify() is False
|
||||
|
||||
def test_default_when_unset_returns_true(self, monkeypatch):
|
||||
"""When CERTIFICATE_VALIDATION is not in config, default is 'enabled'."""
|
||||
import shelfmark.download.network as network
|
||||
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": d)
|
||||
assert network.get_ssl_verify("https://example.com") is True
|
||||
|
||||
|
||||
class TestGetSslVerifyDisabledLocal:
|
||||
"""Tests for 'disabled_local' mode with various address types."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _set_mode(self, monkeypatch):
|
||||
import shelfmark.download.network as network
|
||||
|
||||
self.network = network
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "disabled_local" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
|
||||
# --- Should return False (local addresses) ---
|
||||
|
||||
def test_localhost(self):
|
||||
assert self.network.get_ssl_verify("https://localhost:8080/path") is False
|
||||
|
||||
def test_127_0_0_1(self):
|
||||
assert self.network.get_ssl_verify("http://127.0.0.1:9091") is False
|
||||
|
||||
def test_ipv6_loopback(self):
|
||||
assert self.network.get_ssl_verify("http://[::1]:8080") is False
|
||||
|
||||
def test_private_10_x(self):
|
||||
assert self.network.get_ssl_verify("https://10.0.0.5:443") is False
|
||||
|
||||
def test_private_172_16_x(self):
|
||||
assert self.network.get_ssl_verify("https://172.16.0.1:8080") is False
|
||||
|
||||
def test_private_172_31_x(self):
|
||||
assert self.network.get_ssl_verify("https://172.31.255.255:443") is False
|
||||
|
||||
def test_private_192_168_x(self):
|
||||
assert self.network.get_ssl_verify("https://192.168.1.100:9696") is False
|
||||
|
||||
def test_dot_local_domain(self):
|
||||
assert self.network.get_ssl_verify("https://authelia.local:9091") is False
|
||||
|
||||
def test_dot_internal_domain(self):
|
||||
assert self.network.get_ssl_verify("https://prowlarr.internal:9696") is False
|
||||
|
||||
def test_dot_lan_domain(self):
|
||||
assert self.network.get_ssl_verify("https://server.lan:443") is False
|
||||
|
||||
def test_dot_home_domain(self):
|
||||
assert self.network.get_ssl_verify("https://nas.home:5000") is False
|
||||
|
||||
def test_dot_docker_domain(self):
|
||||
assert self.network.get_ssl_verify("https://app.docker:8080") is False
|
||||
|
||||
def test_simple_hostname_no_dot(self):
|
||||
"""Docker-style service names like 'prowlarr', 'deluge'."""
|
||||
assert self.network.get_ssl_verify("http://prowlarr:9696") is False
|
||||
|
||||
def test_link_local_169_254(self):
|
||||
assert self.network.get_ssl_verify("http://169.254.1.1:8080") is False
|
||||
|
||||
# --- Should return True (public addresses) ---
|
||||
|
||||
def test_public_domain(self):
|
||||
assert self.network.get_ssl_verify("https://example.com") is True
|
||||
|
||||
def test_public_ip(self):
|
||||
assert self.network.get_ssl_verify("https://8.8.8.8:443") is True
|
||||
|
||||
def test_public_subdomain(self):
|
||||
assert self.network.get_ssl_verify("https://api.hardcover.app/v1/graphql") is True
|
||||
|
||||
def test_172_32_is_public(self):
|
||||
"""172.32.x.x is NOT in the private range (only 172.16-31.x.x)."""
|
||||
assert self.network.get_ssl_verify("https://172.32.0.1:443") is True
|
||||
|
||||
def test_empty_url_returns_true(self):
|
||||
"""No URL means we can't determine locality — default to verify."""
|
||||
assert self.network.get_ssl_verify("") is True
|
||||
|
||||
def test_no_url_returns_true(self):
|
||||
assert self.network.get_ssl_verify() is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _apply_ssl_warning_suppression()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplySslWarningSuppression:
|
||||
"""Tests for urllib3 InsecureRequestWarning suppression toggling."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_suppression_flag(self):
|
||||
"""Ensure the module-level flag is clean before each test."""
|
||||
import shelfmark.download.network as network
|
||||
original = network._ssl_warnings_suppressed
|
||||
yield
|
||||
network._ssl_warnings_suppressed = original
|
||||
|
||||
def test_enabled_at_init_is_noop(self, monkeypatch):
|
||||
"""When mode is 'enabled' and warnings were never suppressed, nothing changes."""
|
||||
import shelfmark.download.network as network
|
||||
|
||||
network._ssl_warnings_suppressed = False
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "enabled" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
|
||||
filters_before = list(warnings.filters)
|
||||
network._apply_ssl_warning_suppression()
|
||||
filters_after = list(warnings.filters)
|
||||
|
||||
assert filters_before == filters_after
|
||||
|
||||
def test_disabled_mode_suppresses_warnings(self, monkeypatch):
|
||||
import urllib3
|
||||
import shelfmark.download.network as network
|
||||
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "disabled" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
network._apply_ssl_warning_suppression()
|
||||
|
||||
with warnings.catch_warnings(record=True) as w:
|
||||
warnings.simplefilter("always")
|
||||
warnings.warn("test", urllib3.exceptions.InsecureRequestWarning)
|
||||
|
||||
# urllib3.disable_warnings adds a filter that suppresses — so recorded warnings
|
||||
# should be empty after suppression is applied. However, our catch_warnings
|
||||
# with "always" takes precedence within the context manager. Instead, check
|
||||
# that the filter was installed.
|
||||
filters = [f for f in warnings.filters if len(f) >= 3 and f[2] is urllib3.exceptions.InsecureRequestWarning]
|
||||
assert len(filters) > 0
|
||||
|
||||
def test_disabled_local_mode_suppresses_warnings(self, monkeypatch):
|
||||
import urllib3
|
||||
import shelfmark.download.network as network
|
||||
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "disabled_local" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
network._apply_ssl_warning_suppression()
|
||||
|
||||
filters = [f for f in warnings.filters if len(f) >= 3 and f[2] is urllib3.exceptions.InsecureRequestWarning]
|
||||
assert len(filters) > 0
|
||||
|
||||
def test_enabled_mode_restores_warnings(self, monkeypatch):
|
||||
import urllib3
|
||||
import shelfmark.download.network as network
|
||||
|
||||
# First suppress
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "disabled" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
network._apply_ssl_warning_suppression()
|
||||
|
||||
# Then restore
|
||||
monkeypatch.setattr(network.app_config, "get", lambda k, d="": "enabled" if k == "CERTIFICATE_VALIDATION" else d)
|
||||
network._apply_ssl_warning_suppression()
|
||||
|
||||
# "default" filter should be present for InsecureRequestWarning
|
||||
default_filters = [
|
||||
f for f in warnings.filters
|
||||
if len(f) >= 3 and f[0] == "default" and f[2] is urllib3.exceptions.InsecureRequestWarning
|
||||
]
|
||||
assert len(default_filters) > 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Settings registration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestCertificateValidationSetting:
|
||||
"""Tests for the CERTIFICATE_VALIDATION settings field registration."""
|
||||
|
||||
def _get_network_fields(self):
|
||||
import shelfmark.config.settings # noqa: F401 — ensure settings tabs are registered
|
||||
|
||||
from shelfmark.core.settings_registry import get_settings_tab
|
||||
|
||||
tab = get_settings_tab("network")
|
||||
assert tab is not None
|
||||
return {field.key: field for field in tab.fields if hasattr(field, "key")}
|
||||
|
||||
def test_field_registered(self):
|
||||
fields = self._get_network_fields()
|
||||
assert "CERTIFICATE_VALIDATION" in fields
|
||||
|
||||
def test_field_is_select(self):
|
||||
from shelfmark.core.settings_registry import SelectField
|
||||
|
||||
fields = self._get_network_fields()
|
||||
assert isinstance(fields["CERTIFICATE_VALIDATION"], SelectField)
|
||||
|
||||
def test_field_default_is_enabled(self):
|
||||
fields = self._get_network_fields()
|
||||
assert fields["CERTIFICATE_VALIDATION"].default == "enabled"
|
||||
|
||||
def test_field_has_three_options(self):
|
||||
fields = self._get_network_fields()
|
||||
options = fields["CERTIFICATE_VALIDATION"].options
|
||||
assert len(options) == 3
|
||||
|
||||
def test_field_option_values(self):
|
||||
fields = self._get_network_fields()
|
||||
values = [opt["value"] for opt in fields["CERTIFICATE_VALIDATION"].options]
|
||||
assert values == ["enabled", "disabled_local", "disabled"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Live-apply on settings save
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_update_settings_certificate_validation_triggers_suppression(monkeypatch):
|
||||
"""Changing CERTIFICATE_VALIDATION via update_settings calls _apply_ssl_warning_suppression."""
|
||||
import shelfmark.config.settings # noqa: F401 — ensure settings tabs are registered
|
||||
|
||||
from shelfmark.core.config import config as config_obj
|
||||
from shelfmark.core.settings_registry import update_settings
|
||||
|
||||
monkeypatch.setattr("shelfmark.core.settings_registry.save_config_file", lambda _tab, _values: True)
|
||||
monkeypatch.setattr(config_obj, "refresh", lambda: None)
|
||||
|
||||
called = {"count": 0}
|
||||
|
||||
import shelfmark.download.network as network
|
||||
|
||||
def fake_apply():
|
||||
called["count"] += 1
|
||||
|
||||
monkeypatch.setattr(network, "_apply_ssl_warning_suppression", fake_apply)
|
||||
|
||||
result = update_settings("network", {"CERTIFICATE_VALIDATION": "disabled"})
|
||||
|
||||
assert result["success"] is True
|
||||
assert called["count"] == 1
|
||||
@@ -0,0 +1,115 @@
|
||||
"""SSL verification behavior for download client settings test callbacks."""
|
||||
|
||||
import types
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
def make_config_getter(values):
|
||||
"""Create a config.get function that returns values from a dict."""
|
||||
|
||||
def getter(key, default=""):
|
||||
return values.get(key, default)
|
||||
|
||||
return getter
|
||||
|
||||
|
||||
def test_transmission_settings_test_connection_applies_ssl_verify(monkeypatch):
|
||||
"""Transmission settings callback should apply verify mode to transmission-rpc session."""
|
||||
from shelfmark.core.config import config as config_obj
|
||||
from shelfmark.download.clients import settings as settings_module
|
||||
|
||||
current_values = {
|
||||
"TRANSMISSION_URL": "https://localhost:9091",
|
||||
"TRANSMISSION_USERNAME": "admin",
|
||||
"TRANSMISSION_PASSWORD": "password",
|
||||
}
|
||||
monkeypatch.setattr(config_obj, "get", make_config_getter(current_values))
|
||||
monkeypatch.setattr(settings_module, "get_ssl_verify", lambda _url: False)
|
||||
|
||||
mock_http_session = SimpleNamespace(verify=True)
|
||||
mock_client = MagicMock()
|
||||
mock_client._http_session = mock_http_session
|
||||
mock_client.get_session.return_value = SimpleNamespace(version="4.0.0")
|
||||
|
||||
mock_transmission_rpc = MagicMock()
|
||||
mock_transmission_rpc.Client = MagicMock(return_value=mock_client)
|
||||
|
||||
with patch.dict("sys.modules", {"transmission_rpc": mock_transmission_rpc}):
|
||||
result = settings_module._test_transmission_connection(current_values=current_values)
|
||||
|
||||
assert result["success"] is True
|
||||
assert mock_http_session.verify is False
|
||||
|
||||
|
||||
def test_transmission_settings_test_connection_disables_verify_during_constructor(monkeypatch):
|
||||
"""Settings callback should disable verify before transmission-rpc constructor bootstraps."""
|
||||
from shelfmark.core.config import config as config_obj
|
||||
from shelfmark.download.clients import settings as settings_module
|
||||
|
||||
current_values = {
|
||||
"TRANSMISSION_URL": "https://localhost:9091",
|
||||
"TRANSMISSION_USERNAME": "admin",
|
||||
"TRANSMISSION_PASSWORD": "password",
|
||||
}
|
||||
monkeypatch.setattr(config_obj, "get", make_config_getter(current_values))
|
||||
monkeypatch.setattr(settings_module, "get_ssl_verify", lambda _url: False)
|
||||
|
||||
transmission_pkg = types.ModuleType("transmission_rpc")
|
||||
transmission_pkg.__path__ = []
|
||||
transmission_client_mod = types.ModuleType("transmission_rpc.client")
|
||||
|
||||
def _base_session_factory():
|
||||
return types.SimpleNamespace(verify=True)
|
||||
|
||||
transmission_client_mod.requests = types.SimpleNamespace(Session=_base_session_factory)
|
||||
|
||||
def _fake_client_ctor(**_kwargs):
|
||||
bootstrap_session = transmission_client_mod.requests.Session()
|
||||
if bootstrap_session.verify is not False:
|
||||
raise RuntimeError("verify not disabled during constructor bootstrap")
|
||||
client = MagicMock()
|
||||
client._http_session = bootstrap_session
|
||||
client.get_session.return_value = types.SimpleNamespace(version="4.0.0")
|
||||
return client
|
||||
|
||||
transmission_pkg.Client = _fake_client_ctor
|
||||
transmission_pkg.client = transmission_client_mod
|
||||
|
||||
with patch.dict(
|
||||
"sys.modules",
|
||||
{
|
||||
"transmission_rpc": transmission_pkg,
|
||||
"transmission_rpc.client": transmission_client_mod,
|
||||
},
|
||||
):
|
||||
result = settings_module._test_transmission_connection(current_values=current_values)
|
||||
|
||||
assert result["success"] is True
|
||||
|
||||
|
||||
def test_rtorrent_settings_test_connection_uses_unverified_transport_when_disabled(monkeypatch):
|
||||
"""rTorrent settings callback should pass SafeTransport for HTTPS when verify is disabled."""
|
||||
from shelfmark.core.config import config as config_obj
|
||||
from shelfmark.download.clients import settings as settings_module
|
||||
|
||||
current_values = {
|
||||
"RTORRENT_URL": "https://localhost:8080/RPC2",
|
||||
"RTORRENT_USERNAME": "",
|
||||
"RTORRENT_PASSWORD": "",
|
||||
}
|
||||
monkeypatch.setattr(config_obj, "get", make_config_getter(current_values))
|
||||
monkeypatch.setattr(settings_module, "get_ssl_verify", lambda _url: False)
|
||||
|
||||
mock_rpc = MagicMock()
|
||||
mock_rpc.system.client_version.return_value = "0.9.8"
|
||||
|
||||
mock_xmlrpc = MagicMock()
|
||||
mock_xmlrpc.ServerProxy = MagicMock(return_value=mock_rpc)
|
||||
|
||||
with patch.dict("sys.modules", {"xmlrpc.client": mock_xmlrpc}):
|
||||
result = settings_module._test_rtorrent_connection(current_values=current_values)
|
||||
|
||||
assert result["success"] is True
|
||||
assert mock_xmlrpc.SafeTransport.called is True
|
||||
assert "transport" in mock_xmlrpc.ServerProxy.call_args.kwargs
|
||||
@@ -80,6 +80,36 @@ class TestRTorrentClientIsConfigured:
|
||||
class TestRTorrentClientTestConnection:
|
||||
"""Tests for RTorrentClient.test_connection()."""
|
||||
|
||||
def test_init_https_disabled_verification_uses_unverified_transport(self, monkeypatch):
|
||||
"""HTTPS rTorrent with verify disabled should use a SafeTransport with custom SSL context."""
|
||||
config_values = {
|
||||
"RTORRENT_URL": "https://localhost:8080/RPC2",
|
||||
"RTORRENT_USERNAME": "",
|
||||
"RTORRENT_PASSWORD": "",
|
||||
"RTORRENT_DOWNLOAD_DIR": "/downloads",
|
||||
"RTORRENT_LABEL": "cwabd",
|
||||
}
|
||||
monkeypatch.setattr(
|
||||
"shelfmark.download.clients.rtorrent.config.get",
|
||||
make_config_getter(config_values),
|
||||
)
|
||||
|
||||
mock_rpc = MagicMock()
|
||||
mock_xmlrpc = create_mock_xmlrpc_module()
|
||||
mock_xmlrpc.ServerProxy.return_value = mock_rpc
|
||||
|
||||
with patch.dict("sys.modules", {"xmlrpc.client": mock_xmlrpc}):
|
||||
if "shelfmark.download.clients.rtorrent" in sys.modules:
|
||||
del sys.modules["shelfmark.download.clients.rtorrent"]
|
||||
|
||||
from shelfmark.download.clients import rtorrent as rtorrent_module
|
||||
|
||||
monkeypatch.setattr(rtorrent_module, "get_ssl_verify", lambda _url: False)
|
||||
rtorrent_module.RTorrentClient()
|
||||
|
||||
assert mock_xmlrpc.SafeTransport.called is True
|
||||
assert "transport" in mock_xmlrpc.ServerProxy.call_args.kwargs
|
||||
|
||||
def test_test_connection_success(self, monkeypatch):
|
||||
"""Test successful connection."""
|
||||
config_values = {
|
||||
|
||||
@@ -9,6 +9,7 @@ from unittest.mock import MagicMock, patch
|
||||
from datetime import timedelta
|
||||
import pytest
|
||||
import sys
|
||||
import types
|
||||
|
||||
from shelfmark.download.clients import DownloadStatus
|
||||
|
||||
@@ -154,6 +155,87 @@ class TestTransmissionClientTestConnection:
|
||||
TransmissionClient()
|
||||
assert mock_transmission_rpc.Client.call_args.kwargs.get("protocol") == "https"
|
||||
|
||||
def test_init_applies_certificate_validation_to_session(self, monkeypatch):
|
||||
"""Test Transmission client applies verify mode onto transmission-rpc session."""
|
||||
config_values = {
|
||||
"TRANSMISSION_URL": "https://localhost:9091",
|
||||
"TRANSMISSION_USERNAME": "admin",
|
||||
"TRANSMISSION_PASSWORD": "password",
|
||||
"TRANSMISSION_CATEGORY": "test",
|
||||
}
|
||||
monkeypatch.setattr(
|
||||
"shelfmark.download.clients.transmission.config.get",
|
||||
make_config_getter(config_values),
|
||||
)
|
||||
|
||||
mock_http_session = MagicMock()
|
||||
mock_client_instance = MagicMock()
|
||||
mock_client_instance._http_session = mock_http_session
|
||||
|
||||
mock_transmission_rpc = create_mock_transmission_rpc_module()
|
||||
mock_transmission_rpc.Client.return_value = mock_client_instance
|
||||
|
||||
with patch.dict("sys.modules", {"transmission_rpc": mock_transmission_rpc}):
|
||||
if "shelfmark.download.clients.transmission" in sys.modules:
|
||||
del sys.modules["shelfmark.download.clients.transmission"]
|
||||
|
||||
from shelfmark.download.clients import transmission as transmission_module
|
||||
|
||||
monkeypatch.setattr(transmission_module, "get_ssl_verify", lambda _url: False)
|
||||
transmission_module.TransmissionClient()
|
||||
|
||||
assert mock_http_session.verify is False
|
||||
|
||||
def test_init_disables_verify_before_constructor_bootstrap(self, monkeypatch):
|
||||
"""verify=False must be in place before transmission-rpc constructor bootstraps RPC session."""
|
||||
config_values = {
|
||||
"TRANSMISSION_URL": "https://localhost:9091",
|
||||
"TRANSMISSION_USERNAME": "admin",
|
||||
"TRANSMISSION_PASSWORD": "password",
|
||||
"TRANSMISSION_CATEGORY": "test",
|
||||
}
|
||||
monkeypatch.setattr(
|
||||
"shelfmark.download.clients.transmission.config.get",
|
||||
make_config_getter(config_values),
|
||||
)
|
||||
|
||||
transmission_pkg = types.ModuleType("transmission_rpc")
|
||||
transmission_pkg.__path__ = [] # Mark as package for submodule imports.
|
||||
transmission_client_mod = types.ModuleType("transmission_rpc.client")
|
||||
|
||||
def _base_session_factory():
|
||||
return types.SimpleNamespace(verify=True)
|
||||
|
||||
transmission_client_mod.requests = types.SimpleNamespace(Session=_base_session_factory)
|
||||
|
||||
def _fake_client_ctor(**_kwargs):
|
||||
bootstrap_session = transmission_client_mod.requests.Session()
|
||||
if bootstrap_session.verify is not False:
|
||||
raise RuntimeError("verify not disabled during constructor bootstrap")
|
||||
client = MagicMock()
|
||||
client._http_session = bootstrap_session
|
||||
client.get_session.return_value = MockSession(version="4.0.5")
|
||||
return client
|
||||
|
||||
transmission_pkg.Client = _fake_client_ctor
|
||||
transmission_pkg.client = transmission_client_mod
|
||||
|
||||
with patch.dict(
|
||||
"sys.modules",
|
||||
{
|
||||
"transmission_rpc": transmission_pkg,
|
||||
"transmission_rpc.client": transmission_client_mod,
|
||||
},
|
||||
):
|
||||
if "shelfmark.download.clients.transmission" in sys.modules:
|
||||
del sys.modules["shelfmark.download.clients.transmission"]
|
||||
|
||||
from shelfmark.download.clients import transmission as transmission_module
|
||||
|
||||
monkeypatch.setattr(transmission_module, "get_ssl_verify", lambda _url: False)
|
||||
client = transmission_module.TransmissionClient()
|
||||
assert client._client._http_session.verify is False
|
||||
|
||||
def test_test_connection_success(self, monkeypatch):
|
||||
"""Test successful connection."""
|
||||
config_values = {
|
||||
|
||||
Reference in New Issue
Block a user