wehub-resource-sync
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
1543 lines
57 KiB
Python
1543 lines
57 KiB
Python
"""
|
|
Tests for web/auth/routes.py
|
|
|
|
Tests cover:
|
|
- Login, register, and logout routes
|
|
- CSRF token endpoint
|
|
- Check auth endpoint
|
|
- Change password endpoint
|
|
- Integrity check endpoint
|
|
- Open redirect prevention
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from flask import Flask
|
|
|
|
from local_deep_research.security.password_validator import PasswordValidator
|
|
|
|
|
|
class TestGetCsrfToken:
|
|
"""Tests for /csrf-token endpoint."""
|
|
|
|
def test_returns_csrf_token(self):
|
|
"""Should return CSRF token."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = True
|
|
|
|
with patch("flask_wtf.csrf.generate_csrf") as mock_csrf:
|
|
mock_csrf.return_value = "test_csrf_token_123"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/auth/csrf-token")
|
|
assert response.status_code == 200
|
|
assert response.json["csrf_token"] == "test_csrf_token_123"
|
|
|
|
|
|
class TestLoginPage:
|
|
"""Tests for GET /login endpoint."""
|
|
|
|
def test_renders_login_page(self):
|
|
"""Should render login page for unauthenticated users."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
app.template_folder = "templates" # May need adjustment
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_render.return_value = "Login Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
client.get("/auth/login")
|
|
# Should call render_template
|
|
mock_render.assert_called()
|
|
|
|
def test_redirects_if_already_logged_in(self):
|
|
"""Should redirect to index if user already logged in."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return "Index"
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config:
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.get("/auth/login")
|
|
assert response.status_code == 302
|
|
|
|
|
|
class TestLogin:
|
|
"""Tests for POST /login endpoint."""
|
|
|
|
def test_returns_400_without_username(self):
|
|
"""Should return 400 when username is missing."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_render.return_value = "Login Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/login",
|
|
data={"username": "", "password": "password123"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_400_without_password(self):
|
|
"""Should return 400 when password is missing."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_render.return_value = "Login Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/login",
|
|
data={"username": "testuser", "password": ""},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_401_for_invalid_credentials(self):
|
|
"""Should return 401 for invalid credentials."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.db_manager"
|
|
) as mock_db_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_db_manager.open_user_database.return_value = None
|
|
mock_render.return_value = "Login Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/login",
|
|
data={"username": "testuser", "password": "wrongpassword"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
def test_returns_503_for_database_init_failure_without_lockout(self):
|
|
"""A DatabaseInitializationError should yield 503 and bypass lockout.
|
|
|
|
Regression for the migration-failure path: the user's password is
|
|
valid (we got far enough to attempt schema init), so penalising
|
|
them with a lockout for a server-side configuration problem
|
|
(e.g. world-writable migrations dir) would be wrong.
|
|
"""
|
|
from local_deep_research.database.encrypted_db import (
|
|
DatabaseInitializationError,
|
|
)
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.db_manager"
|
|
) as mock_db_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.get_account_lockout_manager"
|
|
) as mock_lockout_factory,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_db_manager.open_user_database.side_effect = (
|
|
DatabaseInitializationError("boom")
|
|
)
|
|
mock_render.return_value = "Login Page"
|
|
|
|
mock_lockout = mock_lockout_factory.return_value
|
|
mock_lockout.is_locked.return_value = False
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/login",
|
|
data={
|
|
"username": "validuser",
|
|
"password": "correctpassword",
|
|
},
|
|
)
|
|
assert response.status_code == 503
|
|
# Critical: do not punish the user for a server-side bug.
|
|
mock_lockout.record_failure.assert_not_called()
|
|
|
|
|
|
class TestRegisterPage:
|
|
"""Tests for GET /register endpoint."""
|
|
|
|
def test_redirects_when_registrations_disabled(self):
|
|
"""Should redirect to login when registrations are disabled."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config:
|
|
mock_config.return_value = {"allow_registrations": False}
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/auth/register")
|
|
assert response.status_code == 302
|
|
assert "login" in response.location
|
|
|
|
|
|
class TestRegister:
|
|
"""Tests for POST /register endpoint."""
|
|
|
|
def test_returns_400_for_short_username(self):
|
|
"""Should return 400 when username is too short."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_render.return_value = "Register Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/register",
|
|
data={
|
|
"username": "ab", # Too short
|
|
"password": "password123",
|
|
"confirm_password": "password123",
|
|
"acknowledge": "true",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_400_for_invalid_username_chars(self):
|
|
"""Should return 400 when username contains invalid characters."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_render.return_value = "Register Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/register",
|
|
data={
|
|
"username": "test@user!", # Invalid chars
|
|
"password": "password123",
|
|
"confirm_password": "password123",
|
|
"acknowledge": "true",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_400_for_short_password(self):
|
|
"""Should return 400 when password is too short."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_render.return_value = "Register Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/register",
|
|
data={
|
|
"username": "testuser",
|
|
"password": "short", # Too short
|
|
"confirm_password": "short",
|
|
"acknowledge": "true",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_400_for_password_mismatch(self):
|
|
"""Should return 400 when passwords don't match."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_render.return_value = "Register Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/register",
|
|
data={
|
|
"username": "testuser",
|
|
"password": "password123",
|
|
"confirm_password": "different123",
|
|
"acknowledge": "true",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_400_without_acknowledgment(self):
|
|
"""Should return 400 when acknowledgment not provided."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_render.return_value = "Register Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/register",
|
|
data={
|
|
"username": "testuser",
|
|
"password": "password123",
|
|
"confirm_password": "password123",
|
|
# No acknowledge
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
class TestLogout:
|
|
"""Tests for /logout endpoint."""
|
|
|
|
def test_clears_session_on_logout(self):
|
|
"""Should clear session on logout."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch("local_deep_research.web.auth.routes.db_manager"),
|
|
patch("local_deep_research.web.auth.routes.session_manager"),
|
|
patch(
|
|
"local_deep_research.database.session_passwords.session_password_store"
|
|
),
|
|
):
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
sess["session_id"] = "session_123"
|
|
|
|
response = client.post("/auth/logout")
|
|
assert response.status_code == 302
|
|
|
|
with client.session_transaction() as sess:
|
|
assert "username" not in sess
|
|
|
|
def test_redirects_to_login(self):
|
|
"""Should redirect to login after logout."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch("local_deep_research.web.auth.routes.db_manager"),
|
|
patch("local_deep_research.web.auth.routes.session_manager"),
|
|
patch(
|
|
"local_deep_research.database.session_passwords.session_password_store"
|
|
),
|
|
):
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post("/auth/logout")
|
|
assert response.status_code == 302
|
|
assert "login" in response.location
|
|
|
|
def test_unregisters_user_from_scheduler_on_logout(self):
|
|
"""Should call scheduler.unregister_user() during logout."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
mock_scheduler = MagicMock()
|
|
mock_scheduler.is_running = True
|
|
|
|
with (
|
|
patch("local_deep_research.web.auth.routes.db_manager"),
|
|
patch("local_deep_research.web.auth.routes.session_manager"),
|
|
patch(
|
|
"local_deep_research.database.session_passwords.session_password_store"
|
|
),
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler",
|
|
return_value=mock_scheduler,
|
|
),
|
|
):
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
sess["session_id"] = "session_123"
|
|
|
|
client.post("/auth/logout")
|
|
mock_scheduler.unregister_user.assert_called_once_with(
|
|
"testuser"
|
|
)
|
|
|
|
|
|
class TestCheckAuth:
|
|
"""Tests for /check endpoint."""
|
|
|
|
def test_returns_authenticated_true_when_logged_in(self):
|
|
"""Should return authenticated=True when logged in."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.get("/auth/check")
|
|
assert response.status_code == 200
|
|
assert response.json["authenticated"] is True
|
|
assert response.json["username"] == "testuser"
|
|
|
|
def test_returns_authenticated_false_when_not_logged_in(self):
|
|
"""Should return authenticated=False when not logged in."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/auth/check")
|
|
assert response.status_code == 401
|
|
assert response.json["authenticated"] is False
|
|
|
|
|
|
class TestChangePassword:
|
|
"""Tests for /change-password endpoint."""
|
|
|
|
def test_redirects_when_not_logged_in(self):
|
|
"""Should redirect to login when not authenticated."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/auth/change-password")
|
|
assert response.status_code == 302
|
|
assert "login" in response.location
|
|
|
|
def test_returns_400_without_current_password(self):
|
|
"""Should return 400 when current password is missing."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render:
|
|
mock_render.return_value = "Change Password Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.post(
|
|
"/auth/change-password",
|
|
data={
|
|
"current_password": "",
|
|
"new_password": "newpassword123",
|
|
"confirm_password": "newpassword123",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_400_when_passwords_match(self):
|
|
"""Should return 400 when new password is same as current."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render:
|
|
mock_render.return_value = "Change Password Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.post(
|
|
"/auth/change-password",
|
|
data={
|
|
"current_password": "samepassword123",
|
|
"new_password": "samepassword123",
|
|
"confirm_password": "samepassword123",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_successful_password_change(self):
|
|
"""Should destroy ALL sessions and redirect to login on success."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch("local_deep_research.web.auth.routes.db_manager") as mock_db,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.session_manager"
|
|
) as mock_sm,
|
|
patch(
|
|
"local_deep_research.database.session_passwords.session_password_store"
|
|
) as mock_pw_store,
|
|
):
|
|
mock_db.change_password.return_value = True
|
|
mock_render.return_value = "Change Password Page"
|
|
mock_sm.destroy_all_user_sessions.return_value = 2
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.post(
|
|
"/auth/change-password",
|
|
data={
|
|
"current_password": "OldPass123",
|
|
"new_password": "NewStrongP4ss!",
|
|
"confirm_password": "NewStrongP4ss!",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert response.status_code == 302
|
|
assert "login" in response.location
|
|
|
|
with client.session_transaction() as sess:
|
|
assert "username" not in sess
|
|
|
|
mock_db.change_password.assert_called_once_with(
|
|
"testuser", "OldPass123", "NewStrongP4ss!"
|
|
)
|
|
mock_db.close_user_database.assert_called_once_with("testuser")
|
|
|
|
# All sessions for this user should be destroyed
|
|
mock_sm.destroy_all_user_sessions.assert_called_once_with(
|
|
"testuser"
|
|
)
|
|
|
|
# All stored passwords should be cleared
|
|
mock_pw_store.clear_all_for_user.assert_called_once_with(
|
|
"testuser"
|
|
)
|
|
|
|
def test_returns_401_for_wrong_current_password(self):
|
|
"""Should return 401 when current password is incorrect."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch("local_deep_research.web.auth.routes.db_manager") as mock_db,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
):
|
|
mock_db.change_password.return_value = False
|
|
mock_render.return_value = "Change Password Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.post(
|
|
"/auth/change-password",
|
|
data={
|
|
"current_password": "WrongPass123",
|
|
"new_password": "NewStrongP4ss!",
|
|
"confirm_password": "NewStrongP4ss!",
|
|
},
|
|
)
|
|
assert response.status_code == 401
|
|
mock_render.assert_called_with(
|
|
"auth/change_password.html",
|
|
password_requirements=PasswordValidator.get_requirements(),
|
|
)
|
|
|
|
def test_renders_page_when_authenticated(self):
|
|
"""Should render change password page for authenticated user."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render:
|
|
mock_render.return_value = "Change Password Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.get("/auth/change-password")
|
|
assert response.status_code == 200
|
|
mock_render.assert_called_with(
|
|
"auth/change_password.html",
|
|
password_requirements=PasswordValidator.get_requirements(),
|
|
)
|
|
|
|
def test_renders_template_without_missing_dashboard_route(self):
|
|
"""Should render the real template using the root dashboard route."""
|
|
template_dir = (
|
|
Path(__file__).resolve().parents[3]
|
|
/ "src"
|
|
/ "local_deep_research"
|
|
/ "web"
|
|
/ "templates"
|
|
)
|
|
app = Flask(__name__, template_folder=str(template_dir))
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
app.jinja_env.globals["vite_hmr"] = lambda: ""
|
|
app.jinja_env.globals["vite_asset"] = lambda *_args, **_kwargs: ""
|
|
app.jinja_env.globals["csrf_token"] = lambda: "test-csrf-token"
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return "Index"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.get("/auth/change-password")
|
|
|
|
assert response.status_code == 200
|
|
assert b'href="/"' in response.data
|
|
|
|
def test_returns_400_for_weak_new_password(self):
|
|
"""Should return 400 when new password is too weak."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render:
|
|
mock_render.return_value = "Change Password Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.post(
|
|
"/auth/change-password",
|
|
data={
|
|
"current_password": "OldPass123",
|
|
"new_password": "Pass1",
|
|
"confirm_password": "Pass1",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_returns_400_for_mismatched_new_passwords(self):
|
|
"""Should return 400 when new passwords don't match."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render:
|
|
mock_render.return_value = "Change Password Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.post(
|
|
"/auth/change-password",
|
|
data={
|
|
"current_password": "OldPass123",
|
|
"new_password": "NewStrongP4ss!",
|
|
"confirm_password": "DifferentP4ss!",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
class TestIntegrityCheck:
|
|
"""Tests for /integrity-check endpoint."""
|
|
|
|
def test_returns_401_when_not_authenticated(self):
|
|
"""Should return 401 when not authenticated."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.get("/auth/integrity-check")
|
|
assert response.status_code == 401
|
|
|
|
def test_returns_integrity_status(self):
|
|
"""Should return integrity status for authenticated user."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with patch(
|
|
"local_deep_research.web.auth.routes.db_manager"
|
|
) as mock_db_manager:
|
|
mock_db_manager.check_database_integrity.return_value = True
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
response = client.get("/auth/integrity-check")
|
|
assert response.status_code == 200
|
|
assert response.json["username"] == "testuser"
|
|
assert response.json["integrity"] == "valid"
|
|
|
|
|
|
class TestOpenRedirectPrevention:
|
|
"""Tests for open redirect prevention in login."""
|
|
|
|
def test_blocks_external_redirect(self):
|
|
"""Should block redirect to external domain."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return "Index"
|
|
|
|
mock_engine = MagicMock()
|
|
mock_session = MagicMock()
|
|
mock_settings_manager = MagicMock()
|
|
mock_settings_manager.db_version_matches_package.return_value = True
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.db_manager"
|
|
) as mock_db_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.session_manager"
|
|
) as mock_session_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.auth_db_session"
|
|
) as mock_auth_db,
|
|
patch(
|
|
"local_deep_research.database.temp_auth.temp_auth_store"
|
|
) as mock_temp_auth,
|
|
patch(
|
|
"local_deep_research.database.session_passwords.session_password_store"
|
|
),
|
|
patch(
|
|
"local_deep_research.settings.manager.SettingsManager"
|
|
) as mock_settings_cls,
|
|
patch(
|
|
"local_deep_research.database.library_init.initialize_library_for_user"
|
|
),
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler"
|
|
),
|
|
patch("local_deep_research.database.models.ProviderModel"),
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_db_manager.open_user_database.return_value = mock_engine
|
|
mock_db_manager.get_session.return_value = mock_session
|
|
mock_session_manager.create_session.return_value = "session_123"
|
|
# auth_db_session is only called in a background daemon thread
|
|
# (_perform_post_login_tasks), not in the synchronous login handler.
|
|
# MagicMock auto-supports context manager protocol, which is sufficient
|
|
# here since we only test HTTP redirect behavior, not DB interactions.
|
|
mock_auth_db.return_value = MagicMock()
|
|
mock_settings_cls.return_value = mock_settings_manager
|
|
mock_temp_auth.store_auth.return_value = "test_auth_token"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/login?next=https://evil.com/steal",
|
|
data={"username": "testuser", "password": "password123"},
|
|
)
|
|
|
|
# Should redirect to safe URL, not evil.com
|
|
assert response.status_code == 302
|
|
assert "evil.com" not in response.location
|
|
|
|
def test_allows_safe_relative_redirect(self):
|
|
"""Should allow safe relative redirects."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return "Index"
|
|
|
|
@app.route("/dashboard")
|
|
def dashboard():
|
|
return "Dashboard"
|
|
|
|
mock_engine = MagicMock()
|
|
mock_session = MagicMock()
|
|
mock_settings_manager = MagicMock()
|
|
mock_settings_manager.db_version_matches_package.return_value = True
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.db_manager"
|
|
) as mock_db_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.session_manager"
|
|
) as mock_session_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.auth_db_session"
|
|
) as mock_auth_db,
|
|
patch(
|
|
"local_deep_research.database.temp_auth.temp_auth_store"
|
|
) as mock_temp_auth,
|
|
patch(
|
|
"local_deep_research.database.session_passwords.session_password_store"
|
|
),
|
|
patch(
|
|
"local_deep_research.settings.manager.SettingsManager"
|
|
) as mock_settings_cls,
|
|
patch(
|
|
"local_deep_research.database.library_init.initialize_library_for_user"
|
|
),
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler"
|
|
),
|
|
patch("local_deep_research.database.models.ProviderModel"),
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_db_manager.open_user_database.return_value = mock_engine
|
|
mock_db_manager.get_session.return_value = mock_session
|
|
mock_session_manager.create_session.return_value = "session_123"
|
|
# auth_db_session: background-thread-only, auto-CM sufficient (see first test)
|
|
mock_auth_db.return_value = MagicMock()
|
|
mock_settings_cls.return_value = mock_settings_manager
|
|
mock_temp_auth.store_auth.return_value = "test_auth_token"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/login?next=/dashboard",
|
|
data={"username": "testuser", "password": "password123"},
|
|
)
|
|
|
|
# Should redirect to dashboard
|
|
assert response.status_code == 302
|
|
assert "/dashboard" in response.location
|
|
|
|
def test_path_traversal_blocked(self):
|
|
"""Path traversal in next param is blocked by is_safe_redirect_url.
|
|
|
|
The validator detects '..' in the URL path component and rejects
|
|
it, so login falls back to the index page instead of redirecting
|
|
to the traversal target.
|
|
"""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return "Index"
|
|
|
|
mock_engine = MagicMock()
|
|
mock_session = MagicMock()
|
|
mock_settings_manager = MagicMock()
|
|
mock_settings_manager.db_version_matches_package.return_value = True
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.db_manager"
|
|
) as mock_db_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.session_manager"
|
|
) as mock_session_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.auth_db_session"
|
|
) as mock_auth_db,
|
|
patch(
|
|
"local_deep_research.database.temp_auth.temp_auth_store"
|
|
) as mock_temp_auth,
|
|
patch(
|
|
"local_deep_research.database.session_passwords.session_password_store"
|
|
),
|
|
patch(
|
|
"local_deep_research.settings.manager.SettingsManager"
|
|
) as mock_settings_cls,
|
|
patch(
|
|
"local_deep_research.database.library_init.initialize_library_for_user"
|
|
),
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler"
|
|
),
|
|
patch("local_deep_research.database.models.ProviderModel"),
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_db_manager.open_user_database.return_value = mock_engine
|
|
mock_db_manager.get_session.return_value = mock_session
|
|
mock_session_manager.create_session.return_value = "session_123"
|
|
# auth_db_session: background-thread-only, auto-CM sufficient (see first test)
|
|
mock_auth_db.return_value = MagicMock()
|
|
mock_settings_cls.return_value = mock_settings_manager
|
|
mock_temp_auth.store_auth.return_value = "test_auth_token"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/login?next=/../../../etc/passwd",
|
|
data={"username": "testuser", "password": "password123"},
|
|
)
|
|
|
|
# Path traversal is blocked by is_safe_redirect_url,
|
|
# so login succeeds but redirects to index (fallback)
|
|
assert response.status_code == 302
|
|
assert "etc/passwd" not in response.location
|
|
|
|
def test_encoded_path_traversal_blocked(self):
|
|
"""URL-encoded path traversal (%2e%2e) in next param is blocked."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return "Index"
|
|
|
|
mock_engine = MagicMock()
|
|
mock_session = MagicMock()
|
|
mock_settings_manager = MagicMock()
|
|
mock_settings_manager.db_version_matches_package.return_value = True
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.db_manager"
|
|
) as mock_db_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.session_manager"
|
|
) as mock_session_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.auth_db_session"
|
|
) as mock_auth_db,
|
|
patch(
|
|
"local_deep_research.database.temp_auth.temp_auth_store"
|
|
) as mock_temp_auth,
|
|
patch(
|
|
"local_deep_research.database.session_passwords.session_password_store"
|
|
),
|
|
patch(
|
|
"local_deep_research.settings.manager.SettingsManager"
|
|
) as mock_settings_cls,
|
|
patch(
|
|
"local_deep_research.database.library_init.initialize_library_for_user"
|
|
),
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler"
|
|
),
|
|
patch("local_deep_research.database.models.ProviderModel"),
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_db_manager.open_user_database.return_value = mock_engine
|
|
mock_db_manager.get_session.return_value = mock_session
|
|
mock_session_manager.create_session.return_value = "session_123"
|
|
# auth_db_session: background-thread-only, auto-CM sufficient (see first test)
|
|
mock_auth_db.return_value = MagicMock()
|
|
mock_settings_cls.return_value = mock_settings_manager
|
|
mock_temp_auth.store_auth.return_value = "test_auth_token"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/login?next=%2e%2e/admin",
|
|
data={"username": "testuser", "password": "password123"},
|
|
)
|
|
|
|
# Encoded path traversal is blocked
|
|
assert response.status_code == 302
|
|
assert "admin" not in response.location
|
|
|
|
|
|
class TestRedirectPathExtraction:
|
|
"""Tests for defense-in-depth path extraction from redirect URLs.
|
|
|
|
After is_safe_redirect_url validates same-host, the code extracts only
|
|
the path+query+fragment via urljoin+urlparse. These tests verify that
|
|
absolute URLs are correctly converted to relative paths and edge cases
|
|
are handled safely.
|
|
"""
|
|
|
|
def _make_app_and_login(self, next_param):
|
|
"""Helper: create app, register auth blueprint, POST login with next param."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return "Index"
|
|
|
|
mock_engine = MagicMock()
|
|
mock_session = MagicMock()
|
|
mock_settings_manager = MagicMock()
|
|
mock_settings_manager.db_version_matches_package.return_value = True
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.db_manager"
|
|
) as mock_db_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.session_manager"
|
|
) as mock_session_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.auth_db_session"
|
|
) as mock_auth_db,
|
|
patch(
|
|
"local_deep_research.database.temp_auth.temp_auth_store"
|
|
) as mock_temp_auth,
|
|
patch(
|
|
"local_deep_research.database.session_passwords.session_password_store"
|
|
),
|
|
patch(
|
|
"local_deep_research.settings.manager.SettingsManager"
|
|
) as mock_settings_cls,
|
|
patch(
|
|
"local_deep_research.database.library_init.initialize_library_for_user"
|
|
),
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler"
|
|
),
|
|
patch("local_deep_research.database.models.ProviderModel"),
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_db_manager.open_user_database.return_value = mock_engine
|
|
mock_db_manager.get_session.return_value = mock_session
|
|
mock_session_manager.create_session.return_value = "session_123"
|
|
# auth_db_session: background-thread-only, auto-CM sufficient (see first test)
|
|
mock_auth_db.return_value = MagicMock()
|
|
mock_settings_cls.return_value = mock_settings_manager
|
|
mock_temp_auth.store_auth.return_value = "test_auth_token"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
f"/auth/login?next={next_param}",
|
|
data={"username": "testuser", "password": "password123"},
|
|
)
|
|
return response
|
|
|
|
def test_absolute_same_origin_url_extracts_path(self):
|
|
"""Absolute same-origin URL should redirect to path only."""
|
|
response = self._make_app_and_login("http://localhost/dashboard")
|
|
assert response.status_code == 302
|
|
assert response.location.endswith("/dashboard")
|
|
assert "localhost" not in response.location.split("/dashboard")[0]
|
|
|
|
def test_preserves_query_string(self):
|
|
"""Query string in next param should be preserved in redirect."""
|
|
# URL-encode the next value so &page=2 isn't split as a separate param
|
|
from urllib.parse import quote
|
|
|
|
response = self._make_app_and_login(
|
|
quote("/search?q=test&page=2", safe="")
|
|
)
|
|
assert response.status_code == 302
|
|
assert "/search?q=test&page=2" in response.location
|
|
|
|
def test_preserves_fragment(self):
|
|
"""Fragment in next param should be preserved in redirect."""
|
|
response = self._make_app_and_login("/dashboard%23section")
|
|
assert response.status_code == 302
|
|
assert "/dashboard" in response.location
|
|
|
|
def test_absolute_url_with_query_extracts_path_and_query(self):
|
|
"""Absolute URL with query should extract path+query only."""
|
|
from urllib.parse import quote
|
|
|
|
response = self._make_app_and_login(
|
|
quote("http://localhost/research?id=123&tab=sources", safe="")
|
|
)
|
|
assert response.status_code == 302
|
|
assert "/research?id=123&tab=sources" in response.location
|
|
# Must not contain the scheme/host
|
|
assert "http://localhost" not in response.location
|
|
|
|
def test_no_next_param_redirects_to_index(self):
|
|
"""Missing next param should redirect to index."""
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return "Index"
|
|
|
|
mock_engine = MagicMock()
|
|
mock_session = MagicMock()
|
|
mock_settings_manager = MagicMock()
|
|
mock_settings_manager.db_version_matches_package.return_value = True
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.db_manager"
|
|
) as mock_db_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.session_manager"
|
|
) as mock_session_manager,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.auth_db_session"
|
|
) as mock_auth_db,
|
|
patch(
|
|
"local_deep_research.database.temp_auth.temp_auth_store"
|
|
) as mock_temp_auth,
|
|
patch(
|
|
"local_deep_research.database.session_passwords.session_password_store"
|
|
),
|
|
patch(
|
|
"local_deep_research.settings.manager.SettingsManager"
|
|
) as mock_settings_cls,
|
|
patch(
|
|
"local_deep_research.database.library_init.initialize_library_for_user"
|
|
),
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler"
|
|
),
|
|
patch("local_deep_research.database.models.ProviderModel"),
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_db_manager.open_user_database.return_value = mock_engine
|
|
mock_db_manager.get_session.return_value = mock_session
|
|
mock_session_manager.create_session.return_value = "session_123"
|
|
# auth_db_session: background-thread-only, auto-CM sufficient (see first test)
|
|
mock_auth_db.return_value = MagicMock()
|
|
mock_settings_cls.return_value = mock_settings_manager
|
|
mock_temp_auth.store_auth.return_value = "test_auth_token"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/login",
|
|
data={"username": "testuser", "password": "password123"},
|
|
)
|
|
assert response.status_code == 302
|
|
assert response.location.endswith("/")
|
|
|
|
|
|
class TestPostLoginTasksPasswordForwarding:
|
|
"""Tests for _perform_post_login_tasks password forwarding."""
|
|
|
|
def test_post_login_tasks_forwards_password(self):
|
|
"""get_user_db_session must be called with both username AND password."""
|
|
with (
|
|
patch(
|
|
"local_deep_research.settings.manager.SettingsManager"
|
|
) as mock_settings_cls,
|
|
patch(
|
|
"local_deep_research.database.library_init.initialize_library_for_user"
|
|
) as mock_init_lib,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.auth_db_session"
|
|
) as mock_auth_db,
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler"
|
|
),
|
|
patch("local_deep_research.database.models.ProviderModel"),
|
|
patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_get_session,
|
|
):
|
|
mock_settings_mgr = MagicMock()
|
|
mock_settings_mgr.db_version_matches_package.return_value = True
|
|
mock_settings_cls.return_value = mock_settings_mgr
|
|
mock_init_lib.return_value = {"success": True}
|
|
mock_auth_db.return_value.__enter__ = MagicMock(
|
|
return_value=MagicMock()
|
|
)
|
|
mock_auth_db.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
# get_user_db_session is used as context manager
|
|
mock_session = MagicMock()
|
|
mock_get_session.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_get_session.return_value.__exit__ = MagicMock(
|
|
return_value=False
|
|
)
|
|
|
|
from local_deep_research.web.auth.routes import (
|
|
_perform_post_login_tasks,
|
|
)
|
|
|
|
# Call directly (synchronous — @thread_cleanup is transparent)
|
|
_perform_post_login_tasks.__wrapped__("testuser", "testpass")
|
|
|
|
# get_user_db_session should be called with (username, password)
|
|
# at least twice: once for settings migration, once for model cache
|
|
calls = mock_get_session.call_args_list
|
|
password_calls = [
|
|
c
|
|
for c in calls
|
|
if c.args == ("testuser", "testpass")
|
|
or c.kwargs.get("password") == "testpass"
|
|
]
|
|
assert len(password_calls) >= 2, (
|
|
f"Expected get_user_db_session called with password at least "
|
|
f"twice, got {len(password_calls)}: {calls}"
|
|
)
|
|
|
|
|
|
class TestRegistrationTimingAttackPrevention:
|
|
"""Tests for registration timing attack prevention.
|
|
|
|
These tests verify that the registration flow follows OWASP best practices:
|
|
- Generic error messages (prevent content-based enumeration)
|
|
- IntegrityError handling (catches race conditions with generic error + rollback)
|
|
"""
|
|
|
|
def test_generic_error_message_for_duplicate_username(self):
|
|
"""Verify duplicate username returns generic error message (OWASP requirement)."""
|
|
from unittest.mock import patch
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch("local_deep_research.web.auth.routes.db_manager") as mock_db,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
patch("local_deep_research.web.auth.routes.flash") as mock_flash,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_db.user_exists.return_value = True # User already exists
|
|
mock_db.has_encryption = True
|
|
mock_render.return_value = "Register Page"
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/register",
|
|
data={
|
|
"username": "existinguser",
|
|
"password": "TestPassword123!",
|
|
"confirm_password": "TestPassword123!",
|
|
"acknowledge": "true",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
mock_flash.assert_any_call(
|
|
"Registration failed. Please try a different username.",
|
|
"error",
|
|
)
|
|
|
|
def test_integrity_error_returns_generic_message(self):
|
|
"""Verify IntegrityError (race condition) returns generic error (400 not 500)."""
|
|
from unittest.mock import MagicMock, patch
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = "test"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.web.auth.routes.load_server_config"
|
|
) as mock_config,
|
|
patch("local_deep_research.web.auth.routes.db_manager") as mock_db,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.auth_db_session"
|
|
) as mock_auth_db,
|
|
patch(
|
|
"local_deep_research.web.auth.routes.render_template"
|
|
) as mock_render,
|
|
patch("local_deep_research.web.auth.routes.flash") as mock_flash,
|
|
):
|
|
mock_config.return_value = {"allow_registrations": True}
|
|
mock_db.user_exists.return_value = False # Passes pre-check
|
|
mock_db.has_encryption = True
|
|
mock_render.return_value = "Register Page"
|
|
|
|
# Simulate IntegrityError on commit
|
|
mock_session = MagicMock()
|
|
mock_session.add = MagicMock()
|
|
mock_session.commit = MagicMock(
|
|
side_effect=IntegrityError("statement", "params", "orig")
|
|
)
|
|
mock_session.rollback = MagicMock()
|
|
mock_session.close = MagicMock()
|
|
# auth_db_session() is used as a context manager
|
|
mock_auth_db.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_auth_db.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
from local_deep_research.web.auth.routes import auth_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
|
|
with app.test_client() as client:
|
|
response = client.post(
|
|
"/auth/register",
|
|
data={
|
|
"username": "racecondition_user",
|
|
"password": "TestPassword123!",
|
|
"confirm_password": "TestPassword123!",
|
|
"acknowledge": "true",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
mock_flash.assert_any_call(
|
|
"Registration failed. Please try a different username.",
|
|
"error",
|
|
)
|
|
mock_session.rollback.assert_called_once()
|
|
mock_session.add.assert_called_once()
|