7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
547 lines
20 KiB
Python
547 lines
20 KiB
Python
"""
|
|
Tests for web/auth/decorators.py
|
|
|
|
Tests cover:
|
|
- login_required decorator
|
|
- current_user function
|
|
- get_current_db_session function
|
|
- inject_current_user function
|
|
- _safe_redirect_to_login function
|
|
"""
|
|
|
|
from unittest.mock import Mock, patch
|
|
from flask import Blueprint, Flask, session, g
|
|
|
|
|
|
class TestLoginRequiredDecorator:
|
|
"""Tests for login_required decorator."""
|
|
|
|
def test_unauthenticated_api_request_returns_401(self):
|
|
"""Test that unauthenticated API requests return 401 JSON error."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch("local_deep_research.web.auth.decorators.db_manager"):
|
|
from local_deep_research.web.auth.decorators import login_required
|
|
|
|
@app.route("/api/test")
|
|
@login_required
|
|
def test_route():
|
|
return {"success": True}
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/api/test")
|
|
assert response.status_code == 401
|
|
assert response.json["error"] == "Authentication required"
|
|
|
|
def test_unauthenticated_settings_api_request_returns_401(self):
|
|
"""Test that unauthenticated settings API requests return 401."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch("local_deep_research.web.auth.decorators.db_manager"):
|
|
from local_deep_research.web.auth.decorators import login_required
|
|
|
|
@app.route("/settings/api/test")
|
|
@login_required
|
|
def test_route():
|
|
return {"success": True}
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/settings/api/test")
|
|
assert response.status_code == 401
|
|
|
|
def test_unauthenticated_web_request_redirects(self):
|
|
"""Test that unauthenticated web requests redirect to login."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
# Register auth blueprint with login route
|
|
from flask import Blueprint
|
|
|
|
auth = Blueprint("auth", __name__)
|
|
|
|
@auth.route("/login")
|
|
def login():
|
|
return "Login Page"
|
|
|
|
app.register_blueprint(auth, url_prefix="/auth")
|
|
|
|
with patch("local_deep_research.web.auth.decorators.db_manager"):
|
|
from local_deep_research.web.auth.decorators import login_required
|
|
|
|
@app.route("/dashboard")
|
|
@login_required
|
|
def dashboard():
|
|
return "Dashboard"
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/dashboard")
|
|
assert response.status_code == 302
|
|
assert "/auth/login" in response.location
|
|
|
|
def test_authenticated_without_db_connection_api_returns_401(self):
|
|
"""Test authenticated user without DB connection on API route."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.is_user_connected.return_value = False
|
|
|
|
from local_deep_research.web.auth.decorators import login_required
|
|
|
|
@app.route("/api/test")
|
|
@login_required
|
|
def test_route():
|
|
return {"success": True}
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
response = client.get("/api/test")
|
|
assert response.status_code == 401
|
|
assert response.json["error"] == "Database connection required"
|
|
|
|
def test_authenticated_with_db_connection_succeeds(self):
|
|
"""Test authenticated user with DB connection succeeds."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.is_user_connected.return_value = True
|
|
|
|
from local_deep_research.web.auth.decorators import login_required
|
|
|
|
@app.route("/api/test")
|
|
@login_required
|
|
def test_route():
|
|
return {"success": True}
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
response = client.get("/api/test")
|
|
assert response.status_code == 200
|
|
|
|
def test_unauthenticated_nested_news_api_returns_401(self):
|
|
"""Nested API blueprints (e.g. /news/api/...) must return JSON 401,
|
|
not an HTML redirect. Regression guard for the case where API paths
|
|
only matched as a top-level prefix."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch("local_deep_research.web.auth.decorators.db_manager"):
|
|
from local_deep_research.web.auth.decorators import login_required
|
|
|
|
@app.route("/news/api/categories")
|
|
@login_required
|
|
def categories():
|
|
return {"success": True}
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/news/api/categories")
|
|
assert response.status_code == 401
|
|
assert response.json["error"] == "Authentication required"
|
|
|
|
def test_unauthenticated_nested_library_api_returns_401(self):
|
|
"""Nested /library/api/... paths must also return JSON 401."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch("local_deep_research.web.auth.decorators.db_manager"):
|
|
from local_deep_research.web.auth.decorators import login_required
|
|
|
|
@app.route("/library/api/documents")
|
|
@login_required
|
|
def documents():
|
|
return {"success": True}
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/library/api/documents")
|
|
assert response.status_code == 401
|
|
|
|
def test_authenticated_no_db_on_nested_api_returns_401(self):
|
|
"""Stale-session case on a nested API path must return JSON 401,
|
|
not redirect to the login page."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.is_user_connected.return_value = False
|
|
|
|
from local_deep_research.web.auth.decorators import login_required
|
|
|
|
@app.route("/news/api/feed")
|
|
@login_required
|
|
def feed():
|
|
return {"success": True}
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
response = client.get("/news/api/feed")
|
|
assert response.status_code == 401
|
|
assert response.json["error"] == "Database connection required"
|
|
|
|
|
|
class TestIsApiPath:
|
|
"""Tests for the _is_api_path helper."""
|
|
|
|
def test_top_level_api_path(self):
|
|
from local_deep_research.web.auth.decorators import _is_api_path
|
|
|
|
assert _is_api_path("/api/v1/foo") is True
|
|
|
|
def test_nested_news_api_path(self):
|
|
from local_deep_research.web.auth.decorators import _is_api_path
|
|
|
|
assert _is_api_path("/news/api/categories") is True
|
|
|
|
def test_nested_library_api_path(self):
|
|
from local_deep_research.web.auth.decorators import _is_api_path
|
|
|
|
assert _is_api_path("/library/api/documents") is True
|
|
|
|
def test_settings_api_path(self):
|
|
from local_deep_research.web.auth.decorators import _is_api_path
|
|
|
|
assert _is_api_path("/settings/api/foo") is True
|
|
|
|
def test_non_api_page_path(self):
|
|
from local_deep_research.web.auth.decorators import _is_api_path
|
|
|
|
assert _is_api_path("/news/") is False
|
|
assert _is_api_path("/dashboard") is False
|
|
assert _is_api_path("/news/subscriptions") is False
|
|
|
|
def test_partial_api_word_does_not_match(self):
|
|
"""The `api` segment must be slash-bounded — non-API paths whose
|
|
names happen to start with 'api' must NOT be classified as API."""
|
|
from local_deep_research.web.auth.decorators import _is_api_path
|
|
|
|
assert _is_api_path("/apidocs") is False
|
|
assert _is_api_path("/openapi.json") is False
|
|
assert _is_api_path("/notapi/x") is False
|
|
|
|
def test_path_ending_in_slash_api(self):
|
|
"""Paths ending in `/api` (no further segments) are JSON
|
|
endpoints — e.g. /settings/api, /history/api."""
|
|
from local_deep_research.web.auth.decorators import _is_api_path
|
|
|
|
assert _is_api_path("/settings/api") is True
|
|
assert _is_api_path("/history/api") is True
|
|
assert _is_api_path("/foo/api") is True
|
|
assert _is_api_path("/api") is True
|
|
|
|
|
|
class TestCurrentUser:
|
|
"""Tests for current_user function."""
|
|
|
|
def test_current_user_returns_username(self):
|
|
"""Test current_user returns username from session."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
from local_deep_research.web.auth.decorators import current_user
|
|
|
|
with app.test_request_context():
|
|
session["username"] = "testuser"
|
|
assert current_user() == "testuser"
|
|
|
|
def test_current_user_returns_none_when_not_authenticated(self):
|
|
"""Test current_user returns None when not authenticated."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
from local_deep_research.web.auth.decorators import current_user
|
|
|
|
with app.test_request_context():
|
|
assert current_user() is None
|
|
|
|
|
|
class TestGetCurrentDbSession:
|
|
"""Tests for get_current_db_session function."""
|
|
|
|
def test_get_session_returns_session_for_authenticated_user(self):
|
|
"""Test get_current_db_session returns session for authenticated user."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
mock_session = Mock()
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.get_session.return_value = mock_session
|
|
|
|
from local_deep_research.web.auth.decorators import (
|
|
get_current_db_session,
|
|
)
|
|
|
|
with app.test_request_context():
|
|
session["username"] = "testuser"
|
|
result = get_current_db_session()
|
|
assert result == mock_session
|
|
mock_db_manager.get_session.assert_called_once_with("testuser")
|
|
|
|
def test_get_session_returns_none_when_not_authenticated(self):
|
|
"""Test get_current_db_session returns None when not authenticated."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
from local_deep_research.web.auth.decorators import (
|
|
get_current_db_session,
|
|
)
|
|
|
|
with app.test_request_context():
|
|
result = get_current_db_session()
|
|
assert result is None
|
|
|
|
|
|
class TestInjectCurrentUser:
|
|
"""Tests for inject_current_user function."""
|
|
|
|
def test_inject_user_sets_g_current_user(self):
|
|
"""Test inject_current_user sets g.current_user."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
mock_db_session = Mock()
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.get_session.return_value = mock_db_session
|
|
mock_db_manager.is_user_connected.return_value = True
|
|
|
|
from local_deep_research.web.auth.decorators import (
|
|
inject_current_user,
|
|
)
|
|
|
|
with app.test_request_context():
|
|
session["username"] = "testuser"
|
|
inject_current_user()
|
|
assert g.current_user == "testuser"
|
|
# Session is now created lazily, not eagerly in inject_current_user
|
|
assert g.db_session is None
|
|
|
|
def test_inject_user_no_session(self):
|
|
"""Test inject_current_user when no session."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
from local_deep_research.web.auth.decorators import inject_current_user
|
|
|
|
with app.test_request_context():
|
|
inject_current_user()
|
|
assert g.current_user is None
|
|
assert g.db_session is None
|
|
|
|
def test_inject_user_with_db_error(self):
|
|
"""Test inject_current_user handles DB errors gracefully."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.get_session.side_effect = Exception("DB Error")
|
|
|
|
from local_deep_research.web.auth.decorators import (
|
|
inject_current_user,
|
|
)
|
|
|
|
with app.test_request_context():
|
|
session["username"] = "testuser"
|
|
inject_current_user()
|
|
assert g.current_user == "testuser"
|
|
assert g.db_session is None
|
|
|
|
def test_inject_user_clears_stale_session_for_regular_routes(self):
|
|
"""Test inject_current_user clears stale session for regular routes."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.get_session.return_value = None
|
|
mock_db_manager.is_user_connected.return_value = False
|
|
|
|
from local_deep_research.web.auth.decorators import (
|
|
inject_current_user,
|
|
)
|
|
|
|
with app.test_request_context("/dashboard"):
|
|
session["username"] = "testuser"
|
|
inject_current_user()
|
|
# Session should be cleared
|
|
assert g.current_user is None
|
|
assert g.db_session is None
|
|
|
|
def test_inject_user_allows_api_routes_without_db(self):
|
|
"""Test inject_current_user allows API routes without DB connection."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.get_session.return_value = None
|
|
mock_db_manager.is_user_connected.return_value = False
|
|
|
|
from local_deep_research.web.auth.decorators import (
|
|
inject_current_user,
|
|
)
|
|
|
|
with app.test_request_context("/api/test"):
|
|
session["username"] = "testuser"
|
|
inject_current_user()
|
|
# g.current_user should still be set for API routes
|
|
assert g.current_user == "testuser"
|
|
|
|
def test_inject_user_allows_auth_routes_without_db(self):
|
|
"""Test inject_current_user allows auth routes without DB connection."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.get_session.return_value = None
|
|
mock_db_manager.is_user_connected.return_value = False
|
|
|
|
from local_deep_research.web.auth.decorators import (
|
|
inject_current_user,
|
|
)
|
|
|
|
with app.test_request_context("/auth/login"):
|
|
session["username"] = "testuser"
|
|
inject_current_user()
|
|
# g.current_user should still be set for auth routes
|
|
assert g.current_user == "testuser"
|
|
|
|
|
|
class TestSafeRedirectToLogin:
|
|
"""Tests for _safe_redirect_to_login open redirect prevention."""
|
|
|
|
def _setup_app_with_login_required_route(self):
|
|
"""Create a Flask app with auth blueprint and a protected route."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
|
|
auth = Blueprint("auth", __name__)
|
|
|
|
@auth.route("/login")
|
|
def login():
|
|
return "Login Page"
|
|
|
|
app.register_blueprint(auth, url_prefix="/auth")
|
|
return app
|
|
|
|
def test_valid_same_host_next_url_is_preserved(self):
|
|
"""Valid same-host URL should be included as next parameter."""
|
|
app = self._setup_app_with_login_required_route()
|
|
|
|
with patch("local_deep_research.web.auth.decorators.db_manager"):
|
|
from local_deep_research.web.auth.decorators import login_required
|
|
|
|
@app.route("/dashboard")
|
|
@login_required
|
|
def dashboard():
|
|
return "Dashboard"
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/dashboard")
|
|
assert response.status_code == 302
|
|
location = response.location
|
|
assert "/auth/login" in location
|
|
assert "next=" in location
|
|
|
|
def test_protocol_relative_url_rejected(self):
|
|
"""Protocol-relative URLs (//evil.com) should NOT appear in next param."""
|
|
app = self._setup_app_with_login_required_route()
|
|
|
|
with patch("local_deep_research.web.auth.decorators.db_manager"):
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.URLValidator"
|
|
) as mock_validator:
|
|
# Simulate URLValidator rejecting the URL
|
|
mock_validator.is_safe_redirect_url.return_value = False
|
|
|
|
from local_deep_research.web.auth.decorators import (
|
|
_safe_redirect_to_login,
|
|
)
|
|
|
|
with app.test_request_context(
|
|
"//evil.com/steal-cookies",
|
|
base_url="http://localhost:5000",
|
|
):
|
|
response = _safe_redirect_to_login()
|
|
assert response.status_code == 302
|
|
# Should redirect to login WITHOUT next parameter
|
|
assert "next=" not in response.location
|
|
|
|
def test_path_traversal_url_rejected(self):
|
|
"""Path traversal in request URL should NOT appear in next param."""
|
|
app = self._setup_app_with_login_required_route()
|
|
|
|
with patch("local_deep_research.web.auth.decorators.db_manager"):
|
|
from local_deep_research.web.auth.decorators import (
|
|
_safe_redirect_to_login,
|
|
)
|
|
|
|
with app.test_request_context(
|
|
"http://localhost:5000/../../../etc/passwd",
|
|
base_url="http://localhost:5000",
|
|
):
|
|
response = _safe_redirect_to_login()
|
|
assert response.status_code == 302
|
|
assert "next=" not in response.location
|
|
|
|
def test_safe_url_includes_next_param(self):
|
|
"""Safe URLs should include the next parameter."""
|
|
app = self._setup_app_with_login_required_route()
|
|
|
|
with patch("local_deep_research.web.auth.decorators.db_manager"):
|
|
from local_deep_research.web.auth.decorators import (
|
|
_safe_redirect_to_login,
|
|
)
|
|
|
|
with app.test_request_context(
|
|
"/dashboard",
|
|
base_url="http://localhost:5000",
|
|
):
|
|
response = _safe_redirect_to_login()
|
|
assert response.status_code == 302
|
|
assert "next=" in response.location
|
|
|
|
def test_stale_session_redirect_uses_safe_redirect(self):
|
|
"""When DB connection is lost, redirect should use _safe_redirect_to_login."""
|
|
app = self._setup_app_with_login_required_route()
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.is_user_connected.return_value = False
|
|
|
|
from local_deep_research.web.auth.decorators import login_required
|
|
|
|
@app.route("/protected")
|
|
@login_required
|
|
def protected():
|
|
return "Protected"
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
response = client.get("/protected")
|
|
assert response.status_code == 302
|
|
assert "/auth/login" in response.location
|