tim
2f4271314b
Plain HTML, CSS and JavaScript modules, with no build step. Hash routes and relative URLs make it work under any path. The job list is polled: every 2 seconds while a job runs, and less often otherwise.
347 lines
16 KiB
Python
347 lines
16 KiB
Python
"""The web API: login modes, the CSRF header, URL_BASE and the web interface."""
|
|
|
|
import base64
|
|
import hashlib
|
|
import io
|
|
import re
|
|
import time
|
|
|
|
import pytest
|
|
from fake_cwa import FakeCwa
|
|
from itsdangerous import TimestampSigner
|
|
from test_build import PNG
|
|
|
|
from bindery import auth, config, orchestrator, web
|
|
from bindery.store import Store
|
|
|
|
H = {"X-Requested-With": "bindery"}
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(orchestrator, "store", Store(tmp_path / "bindery.db"))
|
|
monkeypatch.setattr(auth, "limiter", auth.LoginLimiter())
|
|
web.app.config["TESTING"] = True
|
|
return web.app.test_client()
|
|
|
|
|
|
@pytest.fixture
|
|
def cwa(monkeypatch):
|
|
server = FakeCwa(username="reader", password="pw")
|
|
monkeypatch.setattr(config, "CWA_URL", server.url)
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "cwa")
|
|
yield server
|
|
server.close()
|
|
|
|
|
|
def login(client, username="reader", password="pw", address="127.0.0.1", forwarded=None):
|
|
headers = {**H, "X-Forwarded-For": forwarded} if forwarded else H
|
|
return client.post(
|
|
"/api/auth/login",
|
|
json={"username": username, "password": password},
|
|
headers=headers,
|
|
environ_base={"REMOTE_ADDR": address},
|
|
)
|
|
|
|
|
|
def test_health_and_no_login_mode(client):
|
|
assert client.get("/api/health").json == {"status": "ok"}
|
|
check = client.get("/api/auth/check").json
|
|
assert check["authenticated"] and check["mode"] == "none"
|
|
assert client.get("/api/stories").json == []
|
|
|
|
|
|
def test_changes_need_the_csrf_header(client):
|
|
assert client.post("/api/stories", json={"urls": []}).status_code == 403
|
|
response = client.post("/api/stories", json={"urls": ["https://example.com/x"]}, headers=H)
|
|
assert response.status_code == 200
|
|
assert response.json["errors"] == ["Not a supported story URL: https://example.com/x"]
|
|
|
|
|
|
def test_errors_are_json(client):
|
|
assert client.get("/api/stories/99").json == {"error": "No story with id 99"}
|
|
assert client.get("/api/stories/99").status_code == 404
|
|
assert client.get("/api/nothing").status_code == 404
|
|
|
|
|
|
def test_cwa_login(client, cwa):
|
|
assert client.get("/api/stories").status_code == 401
|
|
assert client.get("/api/auth/check").json == {
|
|
"authenticated": False,
|
|
"mode": "cwa",
|
|
"user": None,
|
|
"logout_url": None,
|
|
}
|
|
assert login(client, password="wrong").status_code == 401
|
|
assert login(client, username="guest", password="").status_code == 401
|
|
response = login(client)
|
|
assert response.status_code == 200 and response.json["user"] == "reader"
|
|
assert client.get("/api/stories").status_code == 200
|
|
client.post("/api/auth/logout", headers=H)
|
|
assert client.get("/api/stories").status_code == 401
|
|
|
|
|
|
def test_a_session_ends_after_one_day(client, cwa, monkeypatch):
|
|
response = client.post(
|
|
"/api/auth/login", json={"username": "reader", "password": "pw", "remember": True}, headers=H
|
|
)
|
|
assert response.status_code == 200
|
|
assert client.get("/api/stories").status_code == 200
|
|
later = int(time.time()) + 24 * 60 * 60 + 1
|
|
monkeypatch.setattr(TimestampSigner, "get_timestamp", lambda self: later)
|
|
assert client.get("/api/stories").status_code == 401
|
|
|
|
|
|
def test_cwa_login_when_cwa_is_down(client, cwa):
|
|
cwa.close()
|
|
assert login(client).json["error"] == "Cannot check the login with CWA. The log has the details."
|
|
|
|
|
|
def test_cwa_login_lockout_and_allowed_users(client, cwa, monkeypatch):
|
|
for _ in range(auth.MAX_FAILURES):
|
|
login(client, password="wrong")
|
|
assert login(client).json["error"].startswith("Too many wrong passwords")
|
|
monkeypatch.setattr(auth, "limiter", auth.LoginLimiter())
|
|
monkeypatch.setattr(config, "CWA_ALLOWED_USERS", ["someone-else"])
|
|
assert login(client).status_code == 401
|
|
monkeypatch.setattr(config, "CWA_ALLOWED_USERS", ["someone-else", "READER"])
|
|
assert login(client).status_code == 200
|
|
|
|
|
|
def test_wrong_passwords_lock_a_name_only_for_their_address(client, cwa):
|
|
for _ in range(auth.MAX_FAILURES):
|
|
login(client, password="wrong", address="10.0.0.9")
|
|
assert login(client, address="10.0.0.9").json["error"].startswith("Too many wrong passwords")
|
|
assert login(client, address="10.0.0.1").status_code == 200
|
|
|
|
|
|
def test_an_address_has_a_limit_for_all_names(client, cwa):
|
|
for n in range(auth.MAX_ADDRESS_FAILURES):
|
|
login(client, username=f"user{n}", password="wrong", address="10.0.0.9")
|
|
assert login(client, address="10.0.0.9").json["error"].startswith("Too many wrong passwords")
|
|
assert login(client, address="10.0.0.1").status_code == 200
|
|
|
|
|
|
def test_client_address(monkeypatch):
|
|
headers = {"X-Forwarded-For": "1.1.1.1, 10.0.0.9"}
|
|
assert auth.client_address(headers, "172.16.0.2") == "172.16.0.2"
|
|
monkeypatch.setattr(config, "TRUST_PROXY", True)
|
|
assert auth.client_address(headers, "172.16.0.2") == "10.0.0.9"
|
|
assert auth.client_address({}, "172.16.0.2") == "172.16.0.2"
|
|
|
|
|
|
def test_only_a_trusted_proxy_sets_the_client_address(client, cwa, monkeypatch):
|
|
# Without TRUST_PROXY, a client cannot escape the lock with a false header.
|
|
for n in range(auth.MAX_FAILURES):
|
|
login(client, password="wrong", address="10.0.0.9", forwarded=f"1.1.1.{n}")
|
|
assert login(client, address="10.0.0.9", forwarded="1.1.1.9").status_code == 401
|
|
|
|
# With TRUST_PROXY, clients behind the proxy have their own locks. The proxy adds the last entry.
|
|
monkeypatch.setattr(config, "TRUST_PROXY", True)
|
|
monkeypatch.setattr(auth, "limiter", auth.LoginLimiter())
|
|
for n in range(auth.MAX_FAILURES):
|
|
login(client, password="wrong", address="172.16.0.2", forwarded=f"1.1.1.{n}, 10.0.0.9")
|
|
assert login(client, address="172.16.0.2", forwarded="10.0.0.9").status_code == 401
|
|
assert login(client, address="172.16.0.2", forwarded="10.0.0.1").status_code == 200
|
|
|
|
|
|
def test_proxy_header(client, monkeypatch):
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
|
|
monkeypatch.setattr(config, "PROXY_AUTH_LOGOUT_URL", "https://auth.example.com/logout")
|
|
assert client.get("/api/stories").status_code == 401
|
|
assert client.get("/api/stories", headers={"X-Remote-User": "alice"}).status_code == 200
|
|
check = client.get("/api/auth/check", headers={"X-Remote-User": "alice"}).json
|
|
assert check == {
|
|
"authenticated": True,
|
|
"mode": "proxy",
|
|
"user": "alice",
|
|
"logout_url": "https://auth.example.com/logout",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("setting", "value", "error"),
|
|
[
|
|
("AUTH_METHOD", "cwa", "AUTH_METHOD=cwa needs CWA_URL or CWA_DB_PATH"),
|
|
("AUTH_METHOD", "cwa-login", "AUTH_METHOD must be none, cwa or proxy, not 'cwa-login'"),
|
|
("PROXY_AUTH_USER_HEADER", "", "AUTH_METHOD=proxy needs PROXY_AUTH_USER_HEADER"),
|
|
],
|
|
)
|
|
def test_incomplete_login_settings_refuse_every_request(client, monkeypatch, setting, value, error):
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
|
|
monkeypatch.setattr(config, "API_KEY", "k" * 32)
|
|
monkeypatch.setattr(config, setting, value)
|
|
message = f"{error}. Fix the setting and restart Bindery."
|
|
for response in (
|
|
client.get("/api/auth/check"),
|
|
client.get("/api/stories", headers={"X-Remote-User": "alice"}),
|
|
client.get("/api/stories", headers={"X-Api-Key": "k" * 32}),
|
|
client.post("/api/auth/login", json={"username": "a", "password": "b"}, headers=H),
|
|
):
|
|
assert (response.status_code, response.json) == (503, {"error": message})
|
|
assert client.get("/api/health").status_code == 200
|
|
assert client.get("/").status_code == 200
|
|
|
|
|
|
def test_api_key(client, cwa, monkeypatch):
|
|
monkeypatch.setattr(config, "API_KEY", "k" * 32)
|
|
assert client.get("/api/stories", headers={"X-Api-Key": "wrong"}).status_code == 401
|
|
assert client.get("/api/stories", headers={"Authorization": "Bearer " + "k" * 32}).status_code == 200
|
|
# A key needs no CSRF header.
|
|
response = client.post("/api/stories", json={"urls": []}, headers={"X-Api-Key": "k" * 32})
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_epub_upload(client, monkeypatch, tmp_path):
|
|
calls = []
|
|
|
|
def queue(path, name, story_id, check):
|
|
calls.append((path.read_bytes(), name, story_id, check))
|
|
return {"id": "job"}
|
|
|
|
monkeypatch.setattr(orchestrator, "queue_epub_file", queue)
|
|
upload = {"file": (io.BytesIO(b"epub"), "dir/story.epub"), "check": "false"}
|
|
assert client.post("/api/stories/3/import-epub", data=upload, headers=H).json == {"id": "job"}
|
|
assert calls == [(b"epub", "story.epub", 3, False)]
|
|
upload = {"file": (io.BytesIO(b"epub"), "story.epub")}
|
|
assert client.post("/api/stories/import-epub", data=upload, headers=H).status_code == 200
|
|
assert calls[1][2:] == (None, True)
|
|
missing = client.post("/api/stories/import-epub", data={}, headers=H)
|
|
assert (missing.status_code, missing.json) == (400, {"error": "Choose an epub file"})
|
|
assert client.post("/api/stories/import-epub", data={}).status_code == 403
|
|
|
|
|
|
def test_the_cover_of_a_story(client):
|
|
store = orchestrator.store
|
|
story_id = store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", [])
|
|
assert client.get(f"/api/stories/{story_id}").json["story"]["cover"] is None
|
|
assert client.get(f"/api/stories/{story_id}/cover").status_code == 404
|
|
|
|
store.put_image(story_id, "https://example.com/cover.png", "png", "image/png", PNG, is_cover=True)
|
|
version = client.get(f"/api/stories/{story_id}").json["story"]["cover"]
|
|
response = client.get(f"/api/stories/{story_id}/cover?v={version}")
|
|
assert (response.status_code, response.mimetype, response.data) == (200, "image/png", PNG)
|
|
assert response.cache_control.max_age == 365 * 24 * 60 * 60
|
|
|
|
# A new cover gets a new version, so the browser loads it.
|
|
store.put_image(story_id, "https://example.com/new.jpg", "jpg", "image/jpeg", b"new", is_cover=True)
|
|
assert client.get(f"/api/stories/{story_id}").json["story"]["cover"] not in (None, version)
|
|
|
|
# An SVG can hold a script, so Bindery does not serve it.
|
|
store.put_image(story_id, "https://example.com/cover.svg", "svg", "image/svg+xml", b"<svg/>", is_cover=True)
|
|
assert client.get(f"/api/stories/{story_id}").json["story"]["cover"] is None
|
|
assert client.get(f"/api/stories/{story_id}/cover").status_code == 404
|
|
assert client.get("/api/stories/999/cover").status_code == 404
|
|
|
|
|
|
def test_rules_and_the_paragraph_setting(client):
|
|
story_id = orchestrator.store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", [])
|
|
chapter_id = orchestrator.store.add_chapter(story_id, "u1", "One", "<div>a<br>b</div>", "h", "new", 0)
|
|
saved = client.put("/api/rules", json={"rules": [], "paragraphs": True}, headers=H).json
|
|
assert saved == {"rules": [], "paragraphs": True, "global_paragraphs": True}
|
|
story = client.get(f"/api/stories/{story_id}/rules").json
|
|
assert (story["paragraphs"], story["global_paragraphs"]) == (None, True)
|
|
assert orchestrator._paragraphs(story_id)
|
|
client.put(f"/api/stories/{story_id}/rules", json={"rules": [], "paragraphs": False}, headers=H)
|
|
assert not orchestrator._paragraphs(story_id)
|
|
# Rules without the setting keep it.
|
|
client.put(f"/api/stories/{story_id}/rules", json={"rules": []}, headers=H)
|
|
assert client.get(f"/api/stories/{story_id}/rules").json["paragraphs"] is False
|
|
preview = client.post(
|
|
f"/api/stories/{story_id}/rules/preview",
|
|
json={"rules": [], "chapter_id": chapter_id, "source": True},
|
|
headers=H,
|
|
).json
|
|
assert preview["identical"] is False # follows the global setting
|
|
preview = client.post(
|
|
f"/api/stories/{story_id}/rules/preview",
|
|
json={"rules": [], "chapter_id": chapter_id, "paragraphs": False},
|
|
headers=H,
|
|
).json
|
|
assert preview["identical"] is True
|
|
|
|
|
|
def test_delete_revisions(client):
|
|
store = orchestrator.store
|
|
story_id = store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", [])
|
|
chapter_id = store.add_chapter(story_id, "u1", "One", "<p>one</p>", "h1", "new", 0)
|
|
store.add_revision(chapter_id, "<p>two</p>", "h2", "recheck")
|
|
store.add_revision(chapter_id, "<p>three</p>", "h3", "recheck")
|
|
response = client.delete(f"/api/chapters/{chapter_id}/revisions/3", headers=H)
|
|
assert response.status_code == 400 and "in the book" in response.json["error"]
|
|
assert [r["rev"] for r in client.delete(f"/api/chapters/{chapter_id}/revisions/2", headers=H).json] == [1, 3]
|
|
diff = client.get(f"/api/chapters/{chapter_id}/diff").json
|
|
assert (diff["a"], diff["b"], diff["identical"], diff["story_title"]) == (1, 3, False, "T")
|
|
assert client.delete(f"/api/chapters/{chapter_id}/revisions/2", headers=H).status_code == 404
|
|
# The newest revision stays, also when an older revision is in the book.
|
|
client.post(f"/api/chapters/{chapter_id}/revision", json={"rev": 1}, headers=H)
|
|
response = client.delete(f"/api/chapters/{chapter_id}/revisions/3", headers=H)
|
|
assert response.status_code == 400 and "newest text from the site" in response.json["error"]
|
|
assert [r["rev"] for r in client.get(f"/api/chapters/{chapter_id}/diff").json["revisions"]] == [1, 3]
|
|
|
|
|
|
def test_settings(client, monkeypatch, tmp_path):
|
|
ini = tmp_path / "personal.ini"
|
|
ini.write_text("[defaults]\nis_adult:true\n")
|
|
monkeypatch.setattr(config, "PERSONAL_INI", ini)
|
|
assert client.get("/api/settings").json == {
|
|
"recent_chapters": 0,
|
|
"hold_new_links": False,
|
|
"pause_uploads": False,
|
|
"personal_ini": "[defaults]\nis_adult:true\n",
|
|
}
|
|
saved = client.put("/api/settings", json={"recent_chapters": 3}, headers=H).json
|
|
assert (saved["recent_chapters"], saved["warnings"]) == (3, [])
|
|
saved = client.put("/api/settings", json={"hold_new_links": True, "pause_uploads": True}, headers=H).json
|
|
assert (saved["hold_new_links"], saved["pause_uploads"], saved["recent_chapters"]) == (True, True, 3)
|
|
saved = client.put("/api/settings", json={"personal_ini": "[www.notasite.com]\nx:1\n"}, headers=H).json
|
|
assert saved["warnings"] == ["Line 1: Bad Section Name: [www.notasite.com]"]
|
|
assert ini.read_text() == "[www.notasite.com]\nx:1\n"
|
|
bad = client.put("/api/settings", json={"personal_ini": "[defaults\nfoo"}, headers=H)
|
|
assert bad.status_code == 400 and "line 1" in bad.json["error"]
|
|
assert ini.read_text() == "[www.notasite.com]\nx:1\n"
|
|
assert client.put("/api/settings", json={"recent_chapters": -1}, headers=H).status_code == 400
|
|
|
|
|
|
def test_metadata_edit(client):
|
|
story_id = orchestrator.store.create_story(
|
|
"https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", ["title"]
|
|
)
|
|
response = client.put(f"/api/stories/{story_id}/metadata", json={"values": {"authors": "A & B"}}, headers=H)
|
|
assert response.json["story"]["pinned"] == {"authors": ["A", "B"]}
|
|
response = client.put(f"/api/stories/{story_id}/metadata", json={"values": {"title": ""}}, headers=H)
|
|
assert (response.status_code, response.json) == (400, {"error": "A story needs a title"})
|
|
assert client.put(f"/api/stories/{story_id}/metadata", json={}, headers=H).status_code == 400
|
|
assert client.put("/api/stories/999/metadata", json={"values": {}}, headers=H).status_code == 404
|
|
|
|
|
|
def test_web_interface_is_served(client):
|
|
page = client.get("/")
|
|
assert page.status_code == 200 and b'<script type="module" src="static/js/main.js">' in page.data
|
|
assert client.get("/static/js/main.js").status_code == 200
|
|
assert client.get("/static/style.css").status_code == 200
|
|
|
|
|
|
def test_the_page_allows_only_its_own_scripts(client):
|
|
page = client.get("/")
|
|
policy = page.headers["Content-Security-Policy"]
|
|
assert "default-src 'self'" in policy and "unsafe-inline" not in policy
|
|
scripts = re.findall(r"<script>(.*?)</script>", page.get_data(as_text=True), re.DOTALL)
|
|
assert scripts
|
|
for script in scripts:
|
|
digest = base64.b64encode(hashlib.sha256(script.encode()).digest()).decode()
|
|
assert f"'sha256-{digest}'" in policy
|
|
|
|
|
|
def test_url_base():
|
|
app = web.PrefixMiddleware(web.app.wsgi_app, "/bindery")
|
|
web.app.wsgi_app, original = app, web.app.wsgi_app
|
|
try:
|
|
client = web.app.test_client()
|
|
assert client.get("/bindery/api/health").status_code == 200
|
|
assert client.get("/api/health").status_code == 200
|
|
assert client.get("/bindery/static/style.css").status_code == 200
|
|
assert client.get("/other").status_code == 404
|
|
finally:
|
|
web.app.wsgi_app = original
|