234 lines
11 KiB
Python
234 lines
11 KiB
Python
"""One local account, without CWA, with the same login and session protections."""
|
|
|
|
import time
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
from itsdangerous import TimestampSigner
|
|
from test_api import H, login
|
|
from test_api import client as client
|
|
from werkzeug.security import generate_password_hash
|
|
|
|
from bindery import auth, config, orchestrator, web
|
|
|
|
|
|
def password_hash(password="pw", method="pbkdf2:sha256:1000"):
|
|
return generate_password_hash(password, method=method)
|
|
|
|
|
|
@pytest.fixture
|
|
def local(client, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "local")
|
|
monkeypatch.setattr(config, "LOCAL_USERNAME", "reader")
|
|
monkeypatch.setattr(config, "LOCAL_PASSWORD_HASH", password_hash())
|
|
monkeypatch.setattr(config, "CWA_ALLOWED_USERS", ["someone-else"])
|
|
monkeypatch.setattr(config, "CWA_DB_PATH", tmp_path / "missing-app.db")
|
|
monkeypatch.setattr(config, "CALIBRE_DB_PATH", tmp_path / "missing-metadata.db")
|
|
remote = Mock(side_effect=AssertionError("Local logins must not contact CWA"))
|
|
monkeypatch.setattr(orchestrator, "cwa", remote)
|
|
yield remote
|
|
remote.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("method", ["scrypt:1024:8:1", "pbkdf2:sha256:1000", "pbkdf2:sha512:1000", "scrypt"])
|
|
def test_local_login_logout_and_password_hashes(client, local, monkeypatch, method):
|
|
monkeypatch.setattr(config, "LOCAL_PASSWORD_HASH", password_hash(method=method))
|
|
assert client.get("/api/auth/check").json == {
|
|
"authenticated": False,
|
|
"mode": "local",
|
|
"user": None,
|
|
"logout_url": None,
|
|
}
|
|
assert client.get("/api/stories").status_code == 401
|
|
with client.session_transaction() as session:
|
|
session["old_session_data"] = "discard"
|
|
assert login(client).json == {"authenticated": True, "user": "reader"}
|
|
assert client.get("/api/auth/check").json["authenticated"]
|
|
assert client.get("/api/stories").status_code == 200
|
|
with client.session_transaction() as session:
|
|
assert session["mode"] == "local"
|
|
assert session["user"] == "reader"
|
|
assert "old_session_data" not in session
|
|
assert "pw" not in session.values()
|
|
assert config.LOCAL_PASSWORD_HASH not in session.values()
|
|
assert client.post("/api/auth/logout", headers=H).status_code == 200
|
|
assert client.get("/api/stories").status_code == 401
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"username,password", [("reader", "wrong"), ("missing", "pw"), ("READER", "pw"), ("", "pw"), ("reader", "")]
|
|
)
|
|
def test_wrong_credentials_do_not_fall_back(client, local, username, password):
|
|
response = login(client, username=username, password=password)
|
|
assert response.status_code == 401
|
|
assert response.json == {"error": "Wrong user name or password"}
|
|
assert not client.get("/api/auth/check").json["authenticated"]
|
|
|
|
|
|
def test_unicode_user_name_and_password_whitespace(client, local, monkeypatch):
|
|
monkeypatch.setattr(config, "LOCAL_USERNAME", "读者")
|
|
monkeypatch.setattr(config, "LOCAL_PASSWORD_HASH", password_hash(" pässword "))
|
|
assert login(client, username=" 读者 ", password="pässword").status_code == 401
|
|
assert login(client, username=" 读者 ", password=" pässword ").json["user"] == "读者"
|
|
|
|
|
|
def test_wrong_user_name_still_checks_the_password(client, local, monkeypatch):
|
|
assert auth.config_error() is None # Populate the hash validation cache.
|
|
check = Mock(wraps=auth.check_password_hash)
|
|
monkeypatch.setattr(auth, "check_password_hash", check)
|
|
assert login(client, username="missing").status_code == 401
|
|
check.assert_called_once_with(config.LOCAL_PASSWORD_HASH, "pw")
|
|
|
|
|
|
def test_local_user_lockout_expires_and_is_address_specific(client, local, monkeypatch):
|
|
clock = [1000.0]
|
|
monkeypatch.setattr(auth.time, "monotonic", lambda: clock[0])
|
|
for _ in range(auth.MAX_FAILURES):
|
|
assert login(client, password="wrong").status_code == 401
|
|
assert "Too many wrong passwords" in login(client).json["error"]
|
|
assert login(client, address="10.0.0.2").status_code == 200
|
|
clock[0] += auth.LOCKOUT_SECONDS + 1
|
|
assert login(client).status_code == 200
|
|
|
|
|
|
def test_rotating_names_cannot_bypass_the_address_limit(client, local):
|
|
for i in range(auth.MAX_ADDRESS_FAILURES):
|
|
assert login(client, username=f"user-{i}").status_code == 401
|
|
assert "Too many wrong passwords" in login(client).json["error"]
|
|
|
|
|
|
def test_success_clears_failures_and_proxy_addresses_need_trust(client, local, monkeypatch):
|
|
for _ in range(auth.MAX_FAILURES - 1):
|
|
login(client, password="wrong")
|
|
assert login(client).status_code == 200
|
|
assert not auth.limiter._failures["127.0.0.1"]
|
|
for _ in range(auth.MAX_FAILURES):
|
|
login(client, password="wrong")
|
|
assert "Too many wrong passwords" in login(client, forwarded="10.0.0.2").json["error"]
|
|
monkeypatch.setattr(config, "TRUST_PROXY", True)
|
|
assert login(client, forwarded="10.0.0.2").status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"setting,value",
|
|
[
|
|
("LOCAL_USERNAME", ""),
|
|
("LOCAL_PASSWORD_HASH", ""),
|
|
("LOCAL_PASSWORD_HASH", "plaintext"),
|
|
("LOCAL_PASSWORD_HASH", "plain$salt$secret"),
|
|
("LOCAL_PASSWORD_HASH", "scrypt:1024:8:1$$" + "a" * 128),
|
|
("LOCAL_PASSWORD_HASH", "scrypt:3:8:1$salt$" + "a" * 128),
|
|
("LOCAL_PASSWORD_HASH", "pbkdf2:no-such-hash:1000$salt$" + "a" * 64),
|
|
("LOCAL_PASSWORD_HASH", "pbkdf2:sha256:1000$salt$" + "z" * 64),
|
|
("LOCAL_PASSWORD_HASH", "pbkdf2:sha256:1000$salt$abc"),
|
|
("LOCAL_PASSWORD_HASH", "pbkdf2:sha256:999999999999999999999$salt$" + "a" * 64),
|
|
],
|
|
)
|
|
def test_bad_local_configuration_fails_closed(client, local, monkeypatch, setting, value):
|
|
monkeypatch.setattr(config, setting, value)
|
|
monkeypatch.setattr(config, "API_KEY", "test-key")
|
|
for response in [
|
|
client.get("/api/auth/check"),
|
|
client.get("/api/stories", headers={"X-Api-Key": "test-key"}),
|
|
login(client),
|
|
]:
|
|
assert response.status_code == 503
|
|
assert "Fix the setting and restart Bindery" in response.json["error"]
|
|
assert "test-key" not in response.json["error"]
|
|
assert client.get("/api/health").status_code == 200
|
|
|
|
|
|
def test_empty_password_hash_is_not_usable(client, local, monkeypatch):
|
|
monkeypatch.setattr(config, "LOCAL_PASSWORD_HASH", password_hash(""))
|
|
assert login(client, password="").status_code == 503
|
|
with pytest.raises(auth.AuthError, match="nonempty password"):
|
|
auth.login_local("reader", "", "127.0.0.1")
|
|
|
|
|
|
def test_local_login_cannot_be_used_in_another_mode(local, monkeypatch):
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "none")
|
|
with pytest.raises(auth.AuthError, match="does not use local logins"):
|
|
auth.login_local("reader", "pw", "127.0.0.1")
|
|
|
|
|
|
def test_local_logins_and_sessions_keep_csrf_protection(client, local):
|
|
assert client.post("/api/auth/login", json={"username": "reader", "password": "pw"}).status_code == 403
|
|
assert login(client).status_code == 200
|
|
assert client.post("/api/auth/logout").status_code == 403
|
|
assert client.post("/api/stories", json={"urls": []}).status_code == 403
|
|
assert client.post("/api/stories", json={"urls": []}, headers=H).status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize("remember", [False, True])
|
|
def test_remember_me_and_session_expiry(client, local, monkeypatch, remember):
|
|
response = client.post(
|
|
"/api/auth/login", headers=H, json={"username": "reader", "password": "pw", "remember": remember}
|
|
)
|
|
assert response.status_code == 200
|
|
cookie = response.headers["Set-Cookie"]
|
|
assert ("Expires=" in cookie) == remember
|
|
assert "HttpOnly" in cookie and "SameSite=Lax" in cookie
|
|
with client.session_transaction() as session:
|
|
assert session.permanent == remember
|
|
monkeypatch.setattr(TimestampSigner, "get_timestamp", lambda self: int(time.time()) + 86401)
|
|
assert client.get("/api/stories").status_code == 401
|
|
|
|
|
|
@pytest.mark.parametrize("setting,value", [("LOCAL_USERNAME", "renamed"), ("LOCAL_PASSWORD_HASH", None)])
|
|
def test_credential_changes_end_existing_sessions(client, local, monkeypatch, setting, value):
|
|
assert login(client).status_code == 200
|
|
monkeypatch.setattr(config, setting, value or password_hash("new"))
|
|
assert client.get("/api/stories").status_code == 401
|
|
assert login(client, username=config.LOCAL_USERNAME, password="new" if value is None else "pw").status_code == 200
|
|
|
|
|
|
def test_sessions_do_not_cross_login_modes(client, local, monkeypatch):
|
|
assert login(client).status_code == 200
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "cwa")
|
|
monkeypatch.setattr(config, "CWA_ALLOWED_USERS", [])
|
|
assert client.get("/api/stories").status_code == 401
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "local")
|
|
with client.session_transaction() as session:
|
|
session["mode"] = "cwa"
|
|
assert client.get("/api/stories").status_code == 401
|
|
|
|
|
|
def test_api_keys_still_work_in_local_mode(client, local, monkeypatch):
|
|
monkeypatch.setattr(config, "API_KEY", "test-key")
|
|
assert client.get("/api/stories", headers={"X-Api-Key": "test-key"}).status_code == 200
|
|
assert (
|
|
client.post("/api/stories", headers={"Authorization": "Bearer test-key"}, json={"urls": []}).status_code == 200
|
|
)
|
|
assert client.get("/api/stories", headers={"X-Api-Key": "wrong"}).status_code == 401
|
|
|
|
|
|
def test_local_login_under_url_base_uses_a_secure_scoped_cookie(client, local, monkeypatch):
|
|
monkeypatch.setattr(web.app, "wsgi_app", web.PrefixMiddleware(web.app.wsgi_app, "/bindery"))
|
|
monkeypatch.setitem(web.app.config, "SESSION_COOKIE_PATH", "/bindery")
|
|
monkeypatch.setitem(web.app.config, "SESSION_COOKIE_SECURE", True)
|
|
response = client.post(
|
|
"/bindery/api/auth/login",
|
|
base_url="https://localhost",
|
|
headers=H,
|
|
json={"username": "reader", "password": "pw"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "Secure" in response.headers["Set-Cookie"]
|
|
assert "Path=/bindery" in response.headers["Set-Cookie"]
|
|
assert client.get("/bindery/api/stories", base_url="https://localhost").status_code == 200
|
|
assert client.get("/api/stories", base_url="https://localhost").status_code == 404
|
|
|
|
|
|
def test_local_session_needs_its_password_version(client, local):
|
|
assert login(client).status_code == 200
|
|
with client.session_transaction() as session:
|
|
del session["local_password_version"]
|
|
assert client.get("/api/stories").status_code == 401
|
|
|
|
|
|
def test_tampered_cookie_does_not_grant_access(client, local):
|
|
assert login(client).status_code == 200
|
|
cookie = client.get_cookie(web.app.config["SESSION_COOKIE_NAME"])
|
|
client.set_cookie(cookie.key, cookie.value + "changed")
|
|
assert client.get("/api/stories").status_code == 401
|