tim
bb6c87cdd5
BINDERY_ADMIN_USERS lists the admins of the cwa and proxy modes. Only an admin can change the global settings and the global rules, and read or change personal.ini. Without a list, each user is an admin.
544 lines
26 KiB
Python
544 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",
|
|
"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
|