tim
28cd0dcde9
A reader app can browse the stored books by title, author and last change, search them, and download them. The download and cover routes share their code with the story page. An admin turns the catalog on in the settings. The login follows AUTH_METHOD: none needs no login, local and cwa take the account login, and proxy takes the user name from the proxy with an OPDS password that each user sets. The API key does not open the catalog.
549 lines
26 KiB
Python
549 lines
26 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 werkzeug.security import generate_password_hash
|
|
|
|
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,
|
|
"admin": False,
|
|
"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_cwa_that_accepts_every_login_cannot_check_a_password(client, cwa, caplog):
|
|
cwa.anonymous = True
|
|
for password in ("pw", "wrong"):
|
|
response = login(client, password=password)
|
|
assert response.status_code == 401
|
|
assert response.json["error"] == "Cannot check the login with CWA. The log has the details."
|
|
assert "CWA accepts every OPDS login" in caplog.text
|
|
assert client.get("/api/stories").status_code == 401
|
|
# These are not wrong passwords, so they lock no one out.
|
|
cwa.anonymous = False
|
|
orchestrator.cwa()._login_test = None
|
|
assert login(client).status_code == 200
|
|
|
|
|
|
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",
|
|
"admin": True,
|
|
"logout_url": "https://auth.example.com/logout",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(("user", "admin"), [("alice", True), ("Alice", True), ("bob", False)])
|
|
def test_only_an_admin_changes_the_global_settings(client, monkeypatch, tmp_path, user, admin):
|
|
monkeypatch.setattr(config, "PERSONAL_INI", tmp_path / "personal.ini")
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
|
|
monkeypatch.setattr(config, "ADMIN_USERS", ["alice"])
|
|
sid = orchestrator.store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "S", "W", [])
|
|
headers = {**H, "X-Remote-User": user}
|
|
status = 200 if admin else 403
|
|
assert client.get("/api/auth/check", headers=headers).json["admin"] is admin
|
|
assert ("personal_ini" in client.get("/api/settings", headers=headers).json) is admin
|
|
for data in ({"recent_chapters": 5}, {"personal_ini": "[defaults]\n"}):
|
|
assert client.put("/api/settings", json=data, headers=headers).status_code == status
|
|
assert (orchestrator.get_settings()["recent_chapters"] == 5) is admin
|
|
assert config.PERSONAL_INI.exists() is admin
|
|
rules = {"rules": [{"pattern": "a", "replacement": "b"}]}
|
|
response = client.put("/api/rules", json=rules, headers=headers)
|
|
assert response.status_code == status
|
|
assert bool(orchestrator.store.rules(None)) is admin
|
|
if not admin:
|
|
assert response.json == {"error": "Only an admin can do this. BINDERY_ADMIN_USERS lists the admins."}
|
|
# Each user can read the global rules, and change the rules and the settings of a story.
|
|
assert client.get("/api/rules", headers=headers).status_code == 200
|
|
assert client.put(f"/api/stories/{sid}/rules", json=rules, headers=headers).status_code == 200
|
|
assert client.put(f"/api/stories/{sid}/settings", json={"recheck_days": 3}, headers=headers).status_code == 200
|
|
|
|
|
|
def test_the_api_key_and_single_user_modes_are_admins(client, monkeypatch):
|
|
monkeypatch.setattr(config, "ADMIN_USERS", ["alice"])
|
|
assert client.get("/api/auth/check").json["admin"] # the none mode has no users
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
|
|
monkeypatch.setattr(config, "API_KEY", "secret")
|
|
key = {"X-Api-Key": "secret"}
|
|
assert client.get("/api/auth/check", headers=key).json["admin"]
|
|
assert client.put("/api/settings", json={"recent_chapters": 2}, headers=key).status_code == 200
|
|
assert auth.is_admin(auth.Identity("bob", "proxy")) is False
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "local")
|
|
assert auth.is_admin(auth.Identity("bob", "session"))
|
|
# Without a list, each user is an admin.
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
|
|
monkeypatch.setattr(config, "ADMIN_USERS", [])
|
|
assert auth.is_admin(auth.Identity("bob", "proxy"))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("trusted", "address", "accepted"),
|
|
[
|
|
(None, "127.0.0.1", True),
|
|
(None, "::1", True),
|
|
(None, "172.18.0.3", True),
|
|
(None, "192.168.1.20", True),
|
|
(None, "fd00::5", True),
|
|
(None, "::ffff:10.1.2.3", True),
|
|
(None, "203.0.113.9", False),
|
|
(None, "2001:db8::1", False),
|
|
(None, "", False),
|
|
(["172.18.0.3"], "172.18.0.3", True),
|
|
(["172.18.0.3"], "172.18.0.4", False),
|
|
(["172.18.0.3"], "127.0.0.1", False),
|
|
(["10.0.0.0/8", "fd00::/64"], "fd00::7", True),
|
|
(["10.0.0.0/8", "fd00::/64"], "192.168.1.20", False),
|
|
],
|
|
)
|
|
def test_only_a_trusted_address_can_send_the_proxy_header(client, monkeypatch, caplog, trusted, address, accepted):
|
|
monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
|
|
if trusted is not None:
|
|
monkeypatch.setattr(config, "PROXY_AUTH_TRUSTED_IPS", trusted)
|
|
environ = {"REMOTE_ADDR": address}
|
|
response = client.get("/api/stories", headers={"X-Remote-User": "alice"}, environ_base=environ)
|
|
assert response.status_code == (200 if accepted else 401)
|
|
assert (
|
|
client.get("/api/auth/check", headers={"X-Remote-User": "alice"}, environ_base=environ).json["authenticated"]
|
|
is accepted
|
|
)
|
|
assert ("which is not in PROXY_AUTH_TRUSTED_IPS" in caplog.text) is not accepted
|
|
# A change needs the header too, from the same address.
|
|
change = client.post(
|
|
"/api/stories", json={"urls": []}, headers={**H, "X-Remote-User": "alice"}, environ_base=environ
|
|
)
|
|
assert change.status_code == (200 if accepted else 401)
|
|
|
|
|
|
@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, local, cwa or proxy, not 'cwa-login'"),
|
|
("PROXY_AUTH_USER_HEADER", "", "AUTH_METHOD=proxy needs PROXY_AUTH_USER_HEADER"),
|
|
("PROXY_AUTH_TRUSTED_IPS", [], "AUTH_METHOD=proxy needs PROXY_AUTH_TRUSTED_IPS, the addresses of the proxy"),
|
|
(
|
|
"PROXY_AUTH_TRUSTED_IPS",
|
|
["10.0.0.5", "proxy.local"],
|
|
"PROXY_AUTH_TRUSTED_IPS must be a list of addresses and networks: "
|
|
"'proxy.local' does not appear to be an IPv4 or IPv6 network",
|
|
),
|
|
],
|
|
)
|
|
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
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mode", "allowed", "host", "accepted"),
|
|
[
|
|
# Without a login, only an address or a name of a private network.
|
|
("none", [], "localhost", True),
|
|
("none", [], "127.0.0.1:8085", True),
|
|
("none", [], "192.168.1.5:8085", True),
|
|
("none", [], "[fd00::5]:8085", True),
|
|
("none", [], "bindery:8085", True),
|
|
("none", [], "nas.local", True),
|
|
("none", [], "NAS.Home.Arpa.", True),
|
|
("none", [], "bindery.example.com", False),
|
|
("none", [], "rebind.attacker.example:8085", False),
|
|
("none", [], "notlocal", True),
|
|
("none", [], "evil.com.local.attacker.example", False),
|
|
# With a login, each name.
|
|
("local", [], "bindery.example.com", True),
|
|
("proxy", [], "bindery.example.com", True),
|
|
# A list counts in each mode.
|
|
("none", ["bindery.example.com"], "bindery.example.com:443", True),
|
|
("none", ["bindery.example.com"], "localhost", False),
|
|
("none", ["bindery.example.com"], "x.bindery.example.com", False),
|
|
("local", [".example.com"], "bindery.example.com", True),
|
|
("local", [".example.com"], "example.com", True),
|
|
("local", [".example.com"], "badexample.com", False),
|
|
("local", ["example.com"], "attacker.example", False),
|
|
("none", ["*"], "anything.example", True),
|
|
],
|
|
)
|
|
def test_bindery_answers_only_to_its_host_names(client, monkeypatch, mode, allowed, host, accepted):
|
|
monkeypatch.setattr(config, "AUTH_METHOD", mode)
|
|
monkeypatch.setattr(config, "LOCAL_USERNAME", "reader")
|
|
monkeypatch.setattr(config, "LOCAL_PASSWORD_HASH", generate_password_hash("pw", method="pbkdf2:sha256:1000"))
|
|
monkeypatch.setattr(config, "ALLOWED_HOSTS", allowed)
|
|
for path in ("/", "/api/auth/check", "/static/style.css"):
|
|
response = client.get(path, headers={"Host": host})
|
|
assert response.status_code == (200 if accepted else 400), path
|
|
if not accepted:
|
|
assert response.json == {
|
|
"error": f"Bindery does not answer to the host name {host}. Add it to ALLOWED_HOSTS."
|
|
}
|
|
# The health check of the container works under each name.
|
|
assert client.get("/api/health", headers={"Host": host}).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
|
|
assert response.cache_control.private and not response.cache_control.public
|
|
|
|
# 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]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"bad",
|
|
[
|
|
{"recent_chapters": "many"},
|
|
{"main_jobs_count": 100000},
|
|
{"personal_ini": "[defaults\nfoo"},
|
|
{"link_books": "yes"},
|
|
],
|
|
)
|
|
def test_a_bad_setting_changes_no_setting(client, monkeypatch, tmp_path, bad):
|
|
ini = tmp_path / "personal.ini"
|
|
ini.write_text("[defaults]\n")
|
|
monkeypatch.setattr(config, "PERSONAL_INI", ini)
|
|
monkeypatch.setattr(config, "CWA_URL", "https://cwa.invalid")
|
|
before = client.get("/api/settings").json
|
|
good = {
|
|
"link_books": False,
|
|
"hold_new_links": True,
|
|
"pause_uploads": True,
|
|
"recent_chapters": 4,
|
|
"story_jobs_hours": 1,
|
|
"personal_ini": "[defaults]\nis_adult:true\n",
|
|
}
|
|
assert client.put("/api/settings", json={**good, **bad}, headers=H).status_code == 400
|
|
assert client.get("/api/settings").json == before
|
|
|
|
|
|
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)
|
|
monkeypatch.setattr(config, "CWA_URL", "")
|
|
monkeypatch.setattr(config, "CALIBRE_DB_PATH", None)
|
|
assert client.get("/api/settings").json == {
|
|
"recent_chapters": 0,
|
|
"link_books": False,
|
|
"linking_available": False,
|
|
"hold_new_links": False,
|
|
"pause_uploads": False,
|
|
"ingest_available": True,
|
|
"personal_ini": "[defaults]\nis_adult:true\n",
|
|
"opds_enabled": False,
|
|
"opds_password_set": False,
|
|
"opds_user": None,
|
|
"download_template": "{title_sort} - {authors}",
|
|
"download_date_format": "%b, %Y",
|
|
"story_jobs_hours": 36,
|
|
"story_jobs_count": 10,
|
|
"main_jobs_hours": 6,
|
|
"main_jobs_count": 25,
|
|
}
|
|
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
|
|
saved = client.put(
|
|
"/api/settings",
|
|
json={"story_jobs_hours": 48, "story_jobs_count": 20, "main_jobs_hours": 12, "main_jobs_count": 50},
|
|
headers=H,
|
|
).json
|
|
assert (saved["story_jobs_hours"], saved["story_jobs_count"]) == (48, 20)
|
|
assert (saved["main_jobs_hours"], saved["main_jobs_count"]) == (12, 50)
|
|
assert client.put("/api/settings", json={"story_jobs_hours": -1}, headers=H).status_code == 400
|
|
assert client.put("/api/settings", json={"story_jobs_hours": 721}, headers=H).status_code == 400
|
|
assert client.put("/api/settings", json={"main_jobs_count": 501}, 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
|