WebDAV sync support, access tokens, REST control endpoints (#10)
Implement complete WebDAV file serving compatible with various clients from Windows File Explorer to more specialized sync tools. The old control WebSocket has been updated to part-DAV, part REST API instead. Implemented user:pass BASIC auth. Added UI and backend for creating tokens that avoid the need to use actual username and password for requests from CLI or DAV. Reviewed-on: #10
This commit was merged in pull request #10.
This commit is contained in:
@@ -1,53 +0,0 @@
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from cista import config
|
||||
from cista.protocol import Cp, MkDir, Mv, Rename, Rm
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_temp_dir():
|
||||
with tempfile.TemporaryDirectory() as tmpdirname:
|
||||
config.config = config.Config(path=Path(tmpdirname), listen=":0")
|
||||
yield Path(tmpdirname)
|
||||
|
||||
|
||||
def test_mkdir(setup_temp_dir):
|
||||
cmd = MkDir(path="new_folder")
|
||||
cmd()
|
||||
assert (setup_temp_dir / "new_folder").is_dir()
|
||||
|
||||
|
||||
def test_rename(setup_temp_dir):
|
||||
(setup_temp_dir / "old_name").mkdir()
|
||||
cmd = Rename(path="old_name", to="new_name")
|
||||
cmd()
|
||||
assert not (setup_temp_dir / "old_name").exists()
|
||||
assert (setup_temp_dir / "new_name").is_dir()
|
||||
|
||||
|
||||
def test_rm(setup_temp_dir):
|
||||
(setup_temp_dir / "folder_to_remove").mkdir()
|
||||
cmd = Rm(sel=["folder_to_remove"])
|
||||
cmd()
|
||||
assert not (setup_temp_dir / "folder_to_remove").exists()
|
||||
|
||||
|
||||
def test_mv(setup_temp_dir):
|
||||
(setup_temp_dir / "folder_to_move").mkdir()
|
||||
(setup_temp_dir / "destination").mkdir()
|
||||
cmd = Mv(sel=["folder_to_move"], dst="destination")
|
||||
cmd()
|
||||
assert not (setup_temp_dir / "folder_to_move").exists()
|
||||
assert (setup_temp_dir / "destination" / "folder_to_move").is_dir()
|
||||
|
||||
|
||||
def test_cp(setup_temp_dir):
|
||||
(setup_temp_dir / "folder_to_copy").mkdir()
|
||||
(setup_temp_dir / "destination").mkdir()
|
||||
cmd = Cp(sel=["folder_to_copy"], dst="destination")
|
||||
cmd()
|
||||
assert (setup_temp_dir / "folder_to_copy").is_dir()
|
||||
assert (setup_temp_dir / "destination" / "folder_to_copy").is_dir()
|
||||
@@ -0,0 +1,207 @@
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import re
|
||||
import struct
|
||||
from pathlib import Path
|
||||
from time import time
|
||||
from uuid import uuid4
|
||||
|
||||
import jwt
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sanic import Sanic
|
||||
|
||||
from cista import auth, config, session, watching
|
||||
from cista.app import use_session
|
||||
from cista.fileserver import bp as fileserver_bp
|
||||
|
||||
|
||||
def _basic_auth(username: str, password: str) -> dict[str, str]:
|
||||
creds = base64.b64encode(f"{username}:{password}".encode()).decode()
|
||||
return {"Authorization": f"Basic {creds}"}
|
||||
|
||||
|
||||
def _ntlm_type1() -> dict[str, str]:
|
||||
msg = b"NTLMSSP\x00" + struct.pack("<I", 1) + struct.pack("<I", 0x20080205)
|
||||
return {"Authorization": f"NTLM {base64.b64encode(msg).decode()}"}
|
||||
|
||||
|
||||
def _ntlm_type3(username: str, password: str, domain: str, challenge: bytes) -> dict[str, str]:
|
||||
"""Build an NTLMv2 Type 3 message for testing."""
|
||||
from Crypto.Hash import MD4
|
||||
|
||||
# NT hash
|
||||
nt_hash = MD4.new(password.encode("utf-16le")).digest()
|
||||
# NTLMv2 hash
|
||||
ntlmv2_hash = hmac.new(nt_hash, (username.upper() + domain).encode("utf-16le"), hashlib.md5).digest()
|
||||
|
||||
# Build a minimal blob
|
||||
timestamp = struct.pack("<Q", 0)
|
||||
client_nonce = b"\x01" * 8
|
||||
blob = b"\x01\x01\x00\x00\x00\x00\x00\x00" + timestamp + client_nonce + b"\x00\x00\x00\x00"
|
||||
|
||||
# NT proof
|
||||
nt_proof = hmac.new(ntlmv2_hash, challenge + blob, hashlib.md5).digest()
|
||||
nt_response = nt_proof + blob
|
||||
|
||||
domain_enc = domain.encode("utf-16le")
|
||||
username_enc = username.encode("utf-16le")
|
||||
workstation_enc = b""
|
||||
|
||||
lm_response = b"" # Empty for NTLMv2
|
||||
|
||||
# Build Type 3 message
|
||||
msg = bytearray()
|
||||
msg.extend(b"NTLMSSP\x00")
|
||||
msg.extend(struct.pack("<I", 3))
|
||||
|
||||
# Security buffers offsets will be calculated
|
||||
payload_start = 64
|
||||
payloads = []
|
||||
|
||||
def add_buf(data: bytes):
|
||||
offset = payload_start + sum(len(p) for p in payloads)
|
||||
payloads.append(data)
|
||||
return struct.pack("<HHI", len(data), len(data), offset)
|
||||
|
||||
lm_buf = add_buf(lm_response)
|
||||
nt_buf = add_buf(nt_response)
|
||||
domain_buf = add_buf(domain_enc)
|
||||
user_buf = add_buf(username_enc)
|
||||
ws_buf = add_buf(workstation_enc)
|
||||
session_buf = add_buf(b"")
|
||||
|
||||
msg.extend(lm_buf)
|
||||
msg.extend(nt_buf)
|
||||
msg.extend(domain_buf)
|
||||
msg.extend(user_buf)
|
||||
msg.extend(ws_buf)
|
||||
msg.extend(session_buf)
|
||||
msg.extend(struct.pack("<I", 0x20080205))
|
||||
for p in payloads:
|
||||
msg.extend(p)
|
||||
|
||||
return {"Authorization": f"NTLM {base64.b64encode(bytes(msg)).decode()}"}
|
||||
|
||||
|
||||
def _session_cookie_header(username: str) -> dict[str, str]:
|
||||
token = jwt.encode(
|
||||
{"exp": int(time()) + session.max_age, "username": username},
|
||||
session.session_secret(),
|
||||
algorithm="HS256",
|
||||
)
|
||||
return {"Cookie": f"s={token}"}
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_storage(tmp_path: Path):
|
||||
user = config.User()
|
||||
auth.set_password(user, "secret")
|
||||
token = config.Token(key="test_token_123", username="alice")
|
||||
config.config = config.Config(
|
||||
path=tmp_path,
|
||||
listen=":0",
|
||||
public=False,
|
||||
users={"alice": user},
|
||||
tokens={"test_token_123": token},
|
||||
)
|
||||
watching.state.root = []
|
||||
watching.rootpath = tmp_path
|
||||
(tmp_path / "hello.txt").write_text("hello", encoding="utf-8")
|
||||
yield tmp_path
|
||||
watching.state.root = []
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def client(setup_storage: Path):
|
||||
app = Sanic(f"files-auth-test-{uuid4().hex}", strict_slashes=True)
|
||||
app.router.ALLOWED_METHODS = (
|
||||
*app.router.ALLOWED_METHODS,
|
||||
"MKCOL",
|
||||
"MOVE",
|
||||
"COPY",
|
||||
"PROPFIND",
|
||||
)
|
||||
|
||||
@app.on_request
|
||||
async def load_auth_context(request):
|
||||
await use_session(request)
|
||||
|
||||
app.blueprint(fileserver_bp)
|
||||
yield app.asgi_client
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_basic_auth_allows_private_file_access(client):
|
||||
_, res = await client.get("/files/hello.txt", headers=_basic_auth("alice", "secret"))
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.body == b"hello"
|
||||
assert "set-cookie" not in res.headers
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_basic_auth_with_invalid_creds_falls_back_to_session_cookie(client):
|
||||
_, res = await client.get(
|
||||
"/files/hello.txt",
|
||||
headers={**_basic_auth("alice", "wrong"), **_session_cookie_header("alice")},
|
||||
)
|
||||
|
||||
assert res.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_options_unauthenticated_allowed(client):
|
||||
_, res = await client.options("/files/")
|
||||
|
||||
assert res.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unauthenticated_sends_basic_auth_challenge(client):
|
||||
_, res = await client.request("PROPFIND", "/files/")
|
||||
|
||||
assert res.status_code == 401
|
||||
assert res.headers.get("www-authenticate", "").lower().startswith('basic realm="cista"')
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_basic_auth_with_token(client):
|
||||
_, res = await client.get("/files/hello.txt", headers=_basic_auth("token", "test_token_123"))
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.body == b"hello"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_browser_unauthenticated_sends_cookie_challenge(client):
|
||||
_, res = await client.get("/files/", headers={"Accept": "text/html,application/xhtml+xml"})
|
||||
|
||||
assert res.status_code == 401
|
||||
assert res.headers.get("www-authenticate", "").lower().startswith("cookie")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ntlm_auth_with_token(client):
|
||||
# Step 1: request without auth should NOT advertise NTLM
|
||||
# (we prefer clients use BASIC; NTLM still works if client initiates it)
|
||||
_, res1 = await client.get("/files/hello.txt")
|
||||
assert res1.status_code == 401
|
||||
assert "ntlm" not in res1.headers.get("www-authenticate", "").lower()
|
||||
|
||||
# Step 2: client proactively sends Type 1, gets Type 2 challenge
|
||||
_, res2 = await client.get("/files/hello.txt", headers=_ntlm_type1())
|
||||
assert res2.status_code == 401
|
||||
auth_hdr = res2.headers.get("www-authenticate", "")
|
||||
assert auth_hdr.lower().startswith("ntlm ")
|
||||
type2_data = base64.b64decode(auth_hdr.split(" ", 1)[1])
|
||||
challenge = type2_data[24:32]
|
||||
|
||||
# Step 3: send Type 3 with token as password
|
||||
_, res3 = await client.get(
|
||||
"/files/hello.txt",
|
||||
headers=_ntlm_type3("anyuser", "test_token_123", "WORKGROUP", challenge),
|
||||
)
|
||||
assert res3.status_code == 200
|
||||
assert res3.body == b"hello"
|
||||
@@ -0,0 +1,169 @@
|
||||
"""Path traversal and percent-encoding security tests for the fileserver."""
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sanic import Sanic
|
||||
|
||||
from cista import config, watching
|
||||
from cista.fileserver import bp as fileserver_bp
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_storage(tmp_path: Path):
|
||||
config.config = config.Config(path=tmp_path, listen=":0", public=True)
|
||||
watching.state.root = []
|
||||
watching.rootpath = tmp_path
|
||||
yield tmp_path
|
||||
watching.state.root = []
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def client(setup_storage: Path):
|
||||
app = Sanic(f"files-path-sec-test-{uuid4().hex}", strict_slashes=True)
|
||||
app.router.ALLOWED_METHODS = (*app.router.ALLOWED_METHODS, "MKCOL", "MOVE", "COPY", "PROPFIND")
|
||||
app.blueprint(fileserver_bp)
|
||||
yield app.asgi_client
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# %2F — encoded slash should be decoded as a path separator
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_percent2F_decoded_as_path_separator(client, setup_storage: Path):
|
||||
"""%2F in the URL path is decoded to '/' and treated as a path separator."""
|
||||
(setup_storage / "sub").mkdir()
|
||||
(setup_storage / "sub" / "file.txt").write_text("hello", encoding="utf-8")
|
||||
|
||||
_, res = await client.get("/files/sub%2Ffile.txt")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.text == "hello"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_percent2F_creates_nested_directory(client, setup_storage: Path):
|
||||
"""%2F in MKCOL path is decoded as a separator, creating nested dirs."""
|
||||
_, res = await client.request("MKCOL", "/files/parent%2Fchild")
|
||||
|
||||
assert res.status_code == 201
|
||||
assert (setup_storage / "parent" / "child").is_dir()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# %20 — encoded space in filename
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_percent20_in_filename(client, setup_storage: Path):
|
||||
(setup_storage / "my file.txt").write_text("spaced", encoding="utf-8")
|
||||
|
||||
_, res = await client.get("/files/my%20file.txt")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.text == "spaced"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_percent20_in_folder_name(client, setup_storage: Path):
|
||||
_, res = await client.request("MKCOL", "/files/my%20folder")
|
||||
|
||||
assert res.status_code == 201
|
||||
assert (setup_storage / "my folder").is_dir()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Path traversal — .. and encoded variants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_dotdot_rejected(client):
|
||||
""".. is path-normalised by the router before reaching the handler."""
|
||||
_, res = await client.get("/files/..")
|
||||
assert res.status_code in (400, 404)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_dotdot_segment_rejected(client):
|
||||
"""Traversal via sub/../.. is path-normalised by the router."""
|
||||
_, res = await client.get("/files/sub/../..")
|
||||
assert res.status_code in (400, 404)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_encoded_dotdot_rejected(client):
|
||||
"""%2E%2E (encoded ..) must be rejected."""
|
||||
_, res = await client.get("/files/%2E%2E")
|
||||
assert res.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_encoded_dotdot_segment_rejected(client):
|
||||
"""%2E%2E used as a segment in a longer path must be rejected."""
|
||||
_, res = await client.get("/files/sub%2F%2E%2E%2F..%2Fetc%2Fpasswd")
|
||||
assert res.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_dotdot_rejected(client):
|
||||
_, res = await client.request("MKCOL", "/files/..")
|
||||
assert res.status_code in (400, 404)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_dotdot_rejected(client):
|
||||
_, res = await client.delete("/files/..")
|
||||
assert res.status_code in (400, 404)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dot-prefixed filenames (.hidden, ...)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_hidden_file_rejected(client):
|
||||
"""Names starting with '.' are not allowed."""
|
||||
_, res = await client.get("/files/.hidden")
|
||||
assert res.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_hidden_folder_rejected(client):
|
||||
_, res = await client.request("MKCOL", "/files/.secret")
|
||||
assert res.status_code == 400
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Windows-style drive paths (c:/) — safe on Linux, stays inside storage root
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_windows_drive_path_stays_within_root(client, setup_storage: Path):
|
||||
"""A Windows-style drive path like 'c:/foo' is treated as a relative path
|
||||
on Linux and resolves safely inside the storage root."""
|
||||
_, res = await client.request("MKCOL", "/files/c:/secret")
|
||||
|
||||
# Either created inside the storage root (201) or sanitised away (400/404).
|
||||
# The important assertion: nothing was created outside the storage root.
|
||||
assert not (Path("/c:") / "secret").exists()
|
||||
assert not (Path("c:/secret")).exists()
|
||||
if res.status_code == 201:
|
||||
# Created safely inside tmp storage
|
||||
assert (setup_storage / "c:" / "secret").is_dir()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_backslash_in_path_sanitised(client, setup_storage: Path):
|
||||
"""Backslashes are replaced with dashes, not treated as path separators."""
|
||||
_, res = await client.request("MKCOL", "/files/foo\\..\\bar")
|
||||
|
||||
assert res.status_code in (201, 400)
|
||||
# Must not escape storage root
|
||||
assert not (setup_storage.parent / "bar").exists()
|
||||
@@ -0,0 +1,234 @@
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sanic import Sanic
|
||||
|
||||
from cista import config, watching
|
||||
from cista.fileserver import bp as fileserver_bp
|
||||
from cista.protocol import FileEntry
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_storage(tmp_path: Path):
|
||||
config.config = config.Config(path=tmp_path, listen=":0", public=True)
|
||||
watching.state.root = []
|
||||
watching.rootpath = tmp_path
|
||||
yield tmp_path
|
||||
watching.state.root = []
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def client(setup_storage: Path):
|
||||
app = Sanic(f"files-rest-test-{uuid4().hex}", strict_slashes=True)
|
||||
app.router.ALLOWED_METHODS = (*app.router.ALLOWED_METHODS, "MKCOL", "MOVE", "COPY", "PROPFIND")
|
||||
app.blueprint(fileserver_bp)
|
||||
yield app.asgi_client
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_creates_directory(client, setup_storage: Path):
|
||||
_, res = await client.request("MKCOL", "/files/new-folder")
|
||||
|
||||
assert res.status_code == 201
|
||||
assert (setup_storage / "new-folder").is_dir()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_removes_file(client, setup_storage: Path):
|
||||
file_path = setup_storage / "delete-me.txt"
|
||||
file_path.write_text("hello", encoding="utf-8")
|
||||
|
||||
_, res = await client.delete("/files/delete-me.txt")
|
||||
|
||||
assert res.status_code == 204
|
||||
assert not file_path.exists()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_mv_moves_keys_to_target(client, setup_storage: Path):
|
||||
(setup_storage / "target").mkdir()
|
||||
(setup_storage / "alpha.txt").write_text("alpha", encoding="utf-8")
|
||||
(setup_storage / "beta.txt").write_text("beta", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "target", "k-target", 0, 0, 0, 0),
|
||||
FileEntry(1, "alpha.txt", "k-alpha", 0, 5, 0, 1),
|
||||
FileEntry(1, "beta.txt", "k-beta", 0, 4, 0, 1),
|
||||
]
|
||||
|
||||
_, res = await client.post("/files/target?mv=k-alpha+k-beta")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.json["status"] == "ack"
|
||||
assert not (setup_storage / "alpha.txt").exists()
|
||||
assert not (setup_storage / "beta.txt").exists()
|
||||
assert (setup_storage / "target" / "alpha.txt").is_file()
|
||||
assert (setup_storage / "target" / "beta.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_cp_copies_keys_to_target(client, setup_storage: Path):
|
||||
(setup_storage / "target").mkdir()
|
||||
(setup_storage / "copy-me.txt").write_text("copy", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "target", "k-target", 0, 0, 0, 0),
|
||||
FileEntry(1, "copy-me.txt", "k-copy", 0, 4, 0, 1),
|
||||
]
|
||||
|
||||
_, res = await client.post("/files/target?cp=k-copy")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.json["counts"] == {"cp": 1, "mv": 0}
|
||||
assert (setup_storage / "copy-me.txt").is_file()
|
||||
assert (setup_storage / "target" / "copy-me.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_cp_repeated_params_and_plus_form_are_equivalent(
|
||||
client,
|
||||
setup_storage: Path,
|
||||
):
|
||||
(setup_storage / "target").mkdir()
|
||||
(setup_storage / "one.txt").write_text("one", encoding="utf-8")
|
||||
(setup_storage / "two.txt").write_text("two", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "target", "k-target", 0, 0, 0, 0),
|
||||
FileEntry(1, "one.txt", "k-one", 0, 3, 0, 1),
|
||||
FileEntry(1, "two.txt", "k-two", 0, 3, 0, 1),
|
||||
]
|
||||
|
||||
_, res1 = await client.post("/files/target?cp=k-one&cp=k-two")
|
||||
|
||||
assert res1.status_code == 200
|
||||
assert (setup_storage / "target" / "one.txt").is_file()
|
||||
assert (setup_storage / "target" / "two.txt").is_file()
|
||||
|
||||
(setup_storage / "target" / "one.txt").unlink()
|
||||
(setup_storage / "target" / "two.txt").unlink()
|
||||
|
||||
_, res2 = await client.post("/files/target?cp=k-one+k-two")
|
||||
|
||||
assert res2.status_code == 200
|
||||
assert (setup_storage / "target" / "one.txt").is_file()
|
||||
assert (setup_storage / "target" / "two.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_mv_with_to_renames_single_key(
|
||||
client,
|
||||
setup_storage: Path,
|
||||
):
|
||||
(setup_storage / "dst").mkdir()
|
||||
(setup_storage / "old-name.txt").write_text("x", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "dst", "k-dst", 0, 0, 0, 0),
|
||||
FileEntry(1, "old-name.txt", "k-old", 0, 1, 0, 1),
|
||||
]
|
||||
|
||||
_, res = await client.post("/files/dst/new-name.txt?mv=k-old")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert not (setup_storage / "old-name.txt").exists()
|
||||
assert (setup_storage / "dst" / "new-name.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_cp_single_key_to_file_path(client, setup_storage: Path):
|
||||
(setup_storage / "dst").mkdir()
|
||||
(setup_storage / "src.txt").write_text("copy", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "dst", "k-dst", 0, 0, 0, 0),
|
||||
FileEntry(1, "src.txt", "k-src", 0, 4, 0, 1),
|
||||
]
|
||||
|
||||
_, res = await client.post("/files/dst/copied.txt?cp=k-src")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert (setup_storage / "src.txt").is_file()
|
||||
assert (setup_storage / "dst" / "copied.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_supports_combined_cp_then_mv(client, setup_storage: Path):
|
||||
(setup_storage / "target").mkdir()
|
||||
(setup_storage / "copy-me.txt").write_text("copy", encoding="utf-8")
|
||||
(setup_storage / "move-me.txt").write_text("move", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "target", "k-target", 0, 0, 0, 0),
|
||||
FileEntry(1, "copy-me.txt", "k-copy", 0, 4, 0, 1),
|
||||
FileEntry(1, "move-me.txt", "k-move", 0, 4, 0, 1),
|
||||
]
|
||||
|
||||
_, res = await client.post("/files/target?cp=k-copy&mv=k-move")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.json["counts"] == {"cp": 1, "mv": 1}
|
||||
assert (setup_storage / "copy-me.txt").is_file()
|
||||
assert not (setup_storage / "move-me.txt").exists()
|
||||
assert (setup_storage / "target" / "copy-me.txt").is_file()
|
||||
assert (setup_storage / "target" / "move-me.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_rejects_unknown_query_args(client):
|
||||
_, res = await client.post("/files/?cp=k1&wat=1")
|
||||
|
||||
assert res.status_code == 400
|
||||
assert "unknown query parameter" in res.json["message"].lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_requires_query_args(client):
|
||||
_, res = await client.post("/files/")
|
||||
|
||||
assert res.status_code == 400
|
||||
assert "no query arguments" in res.json["message"].lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_rejects_multiple_keys_to_file_target(client, setup_storage: Path):
|
||||
(setup_storage / "a.txt").write_text("a", encoding="utf-8")
|
||||
(setup_storage / "b.txt").write_text("b", encoding="utf-8")
|
||||
(setup_storage / "target.txt").write_text("x", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "a.txt", "k-a", 0, 1, 0, 1),
|
||||
FileEntry(1, "b.txt", "k-b", 0, 1, 0, 1),
|
||||
FileEntry(1, "target.txt", "k-target", 0, 1, 0, 1),
|
||||
]
|
||||
|
||||
_, cp_res = await client.post("/files/target.txt?cp=k-a+k-b")
|
||||
_, mv_res = await client.post("/files/target.txt?mv=k-a+k-b")
|
||||
|
||||
assert cp_res.status_code == 400
|
||||
assert "existing directory" in cp_res.json["message"].lower()
|
||||
assert mv_res.status_code == 400
|
||||
assert "existing directory" in mv_res.json["message"].lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_rejects_directory_to_existing_file_target(client, setup_storage: Path):
|
||||
(setup_storage / "folder").mkdir()
|
||||
(setup_storage / "folder" / "nested.txt").write_text("n", encoding="utf-8")
|
||||
(setup_storage / "existing.txt").write_text("e", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "folder", "k-folder", 0, 0, 0, 0),
|
||||
FileEntry(2, "nested.txt", "k-nested", 0, 1, 0, 1),
|
||||
FileEntry(1, "existing.txt", "k-existing", 0, 1, 0, 1),
|
||||
]
|
||||
|
||||
_, cp_res = await client.post("/files/existing.txt?cp=k-folder")
|
||||
_, mv_res = await client.post("/files/existing.txt?mv=k-folder")
|
||||
|
||||
assert cp_res.status_code == 400
|
||||
assert "directory to an existing file" in cp_res.json["message"].lower()
|
||||
assert mv_res.status_code == 400
|
||||
assert "directory to an existing file" in mv_res.json["message"].lower()
|
||||
@@ -0,0 +1,100 @@
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sanic import Sanic
|
||||
|
||||
from cista import config, watching
|
||||
from cista.fileserver import bp as fileserver_bp
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_storage(tmp_path: Path):
|
||||
config.config = config.Config(path=tmp_path, listen=":0", public=True)
|
||||
watching.state.root = []
|
||||
watching.rootpath = tmp_path
|
||||
yield tmp_path
|
||||
watching.state.root = []
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def client(setup_storage: Path):
|
||||
app = Sanic(f"files-static-test-{uuid4().hex}", strict_slashes=True)
|
||||
app.router.ALLOWED_METHODS = (*app.router.ALLOWED_METHODS, "MKCOL", "MOVE", "COPY", "PROPFIND")
|
||||
app.blueprint(fileserver_bp)
|
||||
yield app.asgi_client
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file_full_content(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.get("/files/hello.txt")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.body == b"hello world"
|
||||
assert res.headers.get("accept-ranges") == "bytes"
|
||||
assert res.headers.get("content-length") == "11"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_head_file_returns_headers_without_body(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.head("/files/hello.txt")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert not res.body
|
||||
assert res.headers.get("content-length") == "11"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file_range_start_end(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.get("/files/hello.txt", headers={"Range": "bytes=1-4"})
|
||||
|
||||
assert res.status_code == 206
|
||||
assert res.body == b"ello"
|
||||
assert res.headers.get("content-range") == "bytes 1-4/11"
|
||||
assert res.headers.get("content-length") == "4"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file_suffix_range(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.get("/files/hello.txt", headers={"Range": "bytes=-5"})
|
||||
|
||||
assert res.status_code == 206
|
||||
assert res.body == b"world"
|
||||
assert res.headers.get("content-range") == "bytes 6-10/11"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_head_file_with_range(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.head("/files/hello.txt", headers={"Range": "bytes=0-4"})
|
||||
|
||||
assert res.status_code == 206
|
||||
assert not res.body
|
||||
assert res.headers.get("content-range") == "bytes 0-4/11"
|
||||
assert res.headers.get("content-length") == "5"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file_unsatisfiable_range_returns_416(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.get("/files/hello.txt", headers={"Range": "bytes=99-100"})
|
||||
|
||||
assert res.status_code == 416
|
||||
assert res.headers.get("content-range") == "bytes */11"
|
||||
@@ -0,0 +1,276 @@
|
||||
"""WebDAV protocol tests: OPTIONS, PROPFIND, PROPPATCH, COPY, MOVE, LOCK, UNLOCK."""
|
||||
import xml.etree.ElementTree as ET
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sanic import Sanic
|
||||
|
||||
from cista import config, watching
|
||||
from cista.fileserver import bp as fileserver_bp
|
||||
from cista.protocol import FileEntry
|
||||
|
||||
_DAV_NS = "DAV:"
|
||||
_METHODS = ("MKCOL", "MOVE", "COPY", "PROPFIND")
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_storage(tmp_path: Path):
|
||||
config.config = config.Config(path=tmp_path, listen=":0", public=True)
|
||||
watching.state.root = []
|
||||
watching.rootpath = tmp_path
|
||||
yield tmp_path
|
||||
watching.state.root = []
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def client(setup_storage: Path):
|
||||
app = Sanic(f"files-dav-test-{uuid4().hex}", strict_slashes=True)
|
||||
app.router.ALLOWED_METHODS = (*app.router.ALLOWED_METHODS, *_METHODS)
|
||||
app.blueprint(fileserver_bp)
|
||||
yield app.asgi_client
|
||||
|
||||
|
||||
def _dav(tag: str) -> str:
|
||||
return f"{{{_DAV_NS}}}{tag}"
|
||||
|
||||
|
||||
def _parse_multistatus(body: bytes) -> list[ET.Element]:
|
||||
root = ET.fromstring(body)
|
||||
assert root.tag == _dav("multistatus")
|
||||
return root.findall(_dav("response"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# OPTIONS
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_options_advertises_dav_class(client):
|
||||
_, res = await client.options("/files/")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert "1" in res.headers.get("dav", "")
|
||||
assert "PROPFIND" in res.headers.get("allow", "")
|
||||
assert "COPY" in res.headers.get("allow", "")
|
||||
assert "MOVE" in res.headers.get("allow", "")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_options_without_trailing_slash(client):
|
||||
"""WebDAV clients (e.g. Windows) send OPTIONS /files without trailing slash."""
|
||||
_, res = await client.options("/files")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert "1" in res.headers.get("dav", "")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PROPFIND
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_propfind_root_depth0(client, setup_storage: Path):
|
||||
_, res = await client.request("PROPFIND", "/files/", headers={"Depth": "0"})
|
||||
|
||||
assert res.status_code == 207
|
||||
responses = _parse_multistatus(res.body)
|
||||
assert len(responses) == 1
|
||||
href = responses[0].findtext(_dav("href"))
|
||||
assert href == "/files/"
|
||||
rt = responses[0].find(f".//{_dav('resourcetype')}/{_dav('collection')}")
|
||||
assert rt is not None, "Root should be a collection"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_propfind_root_depth1_lists_children(client, setup_storage: Path):
|
||||
(setup_storage / "alpha.txt").write_text("a", encoding="utf-8")
|
||||
(setup_storage / "beta").mkdir()
|
||||
|
||||
_, res = await client.request("PROPFIND", "/files/", headers={"Depth": "1"})
|
||||
|
||||
assert res.status_code == 207
|
||||
responses = _parse_multistatus(res.body)
|
||||
hrefs = [r.findtext(_dav("href")) for r in responses]
|
||||
assert "/files/" in hrefs
|
||||
assert "/files/alpha.txt" in hrefs
|
||||
assert "/files/beta/" in hrefs
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_propfind_file_has_content_length(client, setup_storage: Path):
|
||||
(setup_storage / "data.txt").write_text("hello", encoding="utf-8")
|
||||
|
||||
_, res = await client.request("PROPFIND", "/files/data.txt", headers={"Depth": "0"})
|
||||
|
||||
assert res.status_code == 207
|
||||
responses = _parse_multistatus(res.body)
|
||||
cl = responses[0].findtext(f".//{_dav('getcontentlength')}")
|
||||
assert cl == "5"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_propfind_depth_infinity_rejected(client, setup_storage: Path):
|
||||
_, res = await client.request(
|
||||
"PROPFIND", "/files/", headers={"Depth": "infinity"}
|
||||
)
|
||||
assert res.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_propfind_missing_resource_returns_404(client):
|
||||
_, res = await client.request("PROPFIND", "/files/no-such-file.txt")
|
||||
assert res.status_code == 404
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# COPY
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_copy_file_to_new_path(client, setup_storage: Path):
|
||||
(setup_storage / "src.txt").write_text("copy me", encoding="utf-8")
|
||||
|
||||
_, res = await client.request(
|
||||
"COPY",
|
||||
"/files/src.txt",
|
||||
headers={"Destination": "http://localhost/files/dst.txt"},
|
||||
)
|
||||
|
||||
assert res.status_code == 201
|
||||
assert (setup_storage / "src.txt").is_file()
|
||||
assert (setup_storage / "dst.txt").read_text() == "copy me"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_copy_overwrites_existing_by_default(client, setup_storage: Path):
|
||||
(setup_storage / "src.txt").write_text("new", encoding="utf-8")
|
||||
(setup_storage / "dst.txt").write_text("old", encoding="utf-8")
|
||||
|
||||
_, res = await client.request(
|
||||
"COPY",
|
||||
"/files/src.txt",
|
||||
headers={"Destination": "http://localhost/files/dst.txt"},
|
||||
)
|
||||
|
||||
assert res.status_code == 204
|
||||
assert (setup_storage / "dst.txt").read_text() == "new"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_copy_overwrite_false_returns_412(client, setup_storage: Path):
|
||||
(setup_storage / "src.txt").write_text("x", encoding="utf-8")
|
||||
(setup_storage / "dst.txt").write_text("y", encoding="utf-8")
|
||||
|
||||
_, res = await client.request(
|
||||
"COPY",
|
||||
"/files/src.txt",
|
||||
headers={
|
||||
"Destination": "http://localhost/files/dst.txt",
|
||||
"Overwrite": "F",
|
||||
},
|
||||
)
|
||||
|
||||
assert res.status_code == 412
|
||||
assert (setup_storage / "dst.txt").read_text() == "y"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_copy_directory_recursively(client, setup_storage: Path):
|
||||
(setup_storage / "src").mkdir()
|
||||
(setup_storage / "src" / "child.txt").write_text("child", encoding="utf-8")
|
||||
|
||||
_, res = await client.request(
|
||||
"COPY",
|
||||
"/files/src",
|
||||
headers={"Destination": "http://localhost/files/dst"},
|
||||
)
|
||||
|
||||
assert res.status_code == 201
|
||||
assert (setup_storage / "dst" / "child.txt").read_text() == "child"
|
||||
assert (setup_storage / "src" / "child.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_copy_missing_parent_returns_409(client, setup_storage: Path):
|
||||
(setup_storage / "src.txt").write_text("x", encoding="utf-8")
|
||||
|
||||
_, res = await client.request(
|
||||
"COPY",
|
||||
"/files/src.txt",
|
||||
headers={"Destination": "http://localhost/files/nodir/dst.txt"},
|
||||
)
|
||||
|
||||
assert res.status_code == 409
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MOVE
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_move_renames_file(client, setup_storage: Path):
|
||||
(setup_storage / "old.txt").write_text("data", encoding="utf-8")
|
||||
|
||||
_, res = await client.request(
|
||||
"MOVE",
|
||||
"/files/old.txt",
|
||||
headers={"Destination": "http://localhost/files/new.txt"},
|
||||
)
|
||||
|
||||
assert res.status_code == 201
|
||||
assert not (setup_storage / "old.txt").exists()
|
||||
assert (setup_storage / "new.txt").read_text() == "data"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_move_overwrites_existing(client, setup_storage: Path):
|
||||
(setup_storage / "src.txt").write_text("src", encoding="utf-8")
|
||||
(setup_storage / "dst.txt").write_text("dst", encoding="utf-8")
|
||||
|
||||
_, res = await client.request(
|
||||
"MOVE",
|
||||
"/files/src.txt",
|
||||
headers={"Destination": "http://localhost/files/dst.txt"},
|
||||
)
|
||||
|
||||
assert res.status_code == 204
|
||||
assert not (setup_storage / "src.txt").exists()
|
||||
assert (setup_storage / "dst.txt").read_text() == "src"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_move_overwrite_false_returns_412(client, setup_storage: Path):
|
||||
(setup_storage / "src.txt").write_text("src", encoding="utf-8")
|
||||
(setup_storage / "dst.txt").write_text("dst", encoding="utf-8")
|
||||
|
||||
_, res = await client.request(
|
||||
"MOVE",
|
||||
"/files/src.txt",
|
||||
headers={
|
||||
"Destination": "http://localhost/files/dst.txt",
|
||||
"Overwrite": "F",
|
||||
},
|
||||
)
|
||||
|
||||
assert res.status_code == 412
|
||||
assert (setup_storage / "src.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_move_same_source_and_dest_is_noop(client, setup_storage: Path):
|
||||
(setup_storage / "file.txt").write_text("x", encoding="utf-8")
|
||||
|
||||
_, res = await client.request(
|
||||
"MOVE",
|
||||
"/files/file.txt",
|
||||
headers={"Destination": "http://localhost/files/file.txt"},
|
||||
)
|
||||
|
||||
assert res.status_code == 204
|
||||
assert (setup_storage / "file.txt").is_file()
|
||||
@@ -0,0 +1,192 @@
|
||||
from pathlib import Path
|
||||
from time import time
|
||||
from uuid import uuid4
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sanic import Sanic
|
||||
|
||||
from cista import auth, config, watching
|
||||
from cista.auth import bp as auth_bp
|
||||
|
||||
|
||||
def _persist_config():
|
||||
import msgspec
|
||||
from pathlib import PurePath
|
||||
|
||||
def enc_hook(obj):
|
||||
if isinstance(obj, PurePath):
|
||||
return obj.as_posix()
|
||||
raise TypeError
|
||||
|
||||
raw = msgspec.to_builtins(config.config, enc_hook=enc_hook)
|
||||
config.conffile.write_bytes(msgspec.toml.encode(raw))
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_storage(tmp_path: Path):
|
||||
os.environ["CISTA_HOME"] = str(tmp_path)
|
||||
config.init_confdir()
|
||||
user = config.User()
|
||||
auth.set_password(user, "secret")
|
||||
admin = config.User(privileged=True)
|
||||
auth.set_password(admin, "secret")
|
||||
config.config = config.Config(
|
||||
path=tmp_path,
|
||||
listen=":0",
|
||||
public=False,
|
||||
users={"alice": user, "admin": admin},
|
||||
)
|
||||
_persist_config()
|
||||
watching.state.root = []
|
||||
watching.rootpath = tmp_path
|
||||
yield tmp_path
|
||||
watching.state.root = []
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def client(setup_storage: Path):
|
||||
app = Sanic(f"token-test-{uuid4().hex}", strict_slashes=True)
|
||||
app.router.ALLOWED_METHODS = (
|
||||
*app.router.ALLOWED_METHODS,
|
||||
"MKCOL",
|
||||
"MOVE",
|
||||
"COPY",
|
||||
"PROPFIND",
|
||||
)
|
||||
app.blueprint(auth_bp)
|
||||
yield app.asgi_client
|
||||
|
||||
|
||||
def _basic_auth(username: str, password: str) -> str:
|
||||
return f"Basic {__import__('base64').b64encode(f'{username}:{password}'.encode()).decode()}"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_token_crud(client):
|
||||
# Admin creates a token without specifying username (auto-assigned)
|
||||
_, res = await client.post(
|
||||
"/auth/tokens",
|
||||
json={"name": "test"},
|
||||
headers={"Authorization": _basic_auth("admin", "secret")},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
data = res.json
|
||||
assert "id" in data
|
||||
assert "key" in data
|
||||
assert data["username"] == "admin"
|
||||
assert data["name"] == "test"
|
||||
token_id = data["id"]
|
||||
token_key = data["key"]
|
||||
|
||||
# List tokens - admin sees only their own
|
||||
_, res = await client.get(
|
||||
"/auth/tokens",
|
||||
headers={"Authorization": _basic_auth("admin", "secret")},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
tokens = res.json["tokens"]
|
||||
assert len(tokens) == 1
|
||||
assert tokens[0]["id"] == token_id
|
||||
assert tokens[0]["username"] == "admin"
|
||||
|
||||
# Use token via Basic auth (token:<secret>)
|
||||
_, res = await client.get(
|
||||
"/auth/tokens",
|
||||
headers={"Authorization": _basic_auth("token", token_key)},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
|
||||
# Delete token
|
||||
_, res = await client.delete(
|
||||
f"/auth/tokens/{token_id}",
|
||||
headers={"Authorization": _basic_auth("admin", "secret")},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
|
||||
# List should be empty
|
||||
_, res = await client.get(
|
||||
"/auth/tokens",
|
||||
headers={"Authorization": _basic_auth("admin", "secret")},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
assert len(res.json["tokens"]) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_token_user_scoped(client):
|
||||
# Alice creates a token for herself (no username specified)
|
||||
_, res = await client.post(
|
||||
"/auth/tokens",
|
||||
json={"name": "alice-token"},
|
||||
headers={"Authorization": _basic_auth("alice", "secret")},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
alice_token_id = res.json["id"]
|
||||
alice_token_key = res.json["key"]
|
||||
|
||||
# Admin creates a token for themselves
|
||||
_, res = await client.post(
|
||||
"/auth/tokens",
|
||||
json={"name": "admin-token"},
|
||||
headers={"Authorization": _basic_auth("admin", "secret")},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
admin_token_id = res.json["id"]
|
||||
|
||||
# Alice lists tokens - sees only her own
|
||||
_, res = await client.get(
|
||||
"/auth/tokens",
|
||||
headers={"Authorization": _basic_auth("alice", "secret")},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
tokens = res.json["tokens"]
|
||||
assert len(tokens) == 1
|
||||
assert tokens[0]["id"] == alice_token_id
|
||||
assert tokens[0]["username"] == "alice"
|
||||
|
||||
# Admin lists tokens - sees only their own
|
||||
_, res = await client.get(
|
||||
"/auth/tokens",
|
||||
headers={"Authorization": _basic_auth("admin", "secret")},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
tokens = res.json["tokens"]
|
||||
assert len(tokens) == 1
|
||||
assert tokens[0]["id"] == admin_token_id
|
||||
assert tokens[0]["username"] == "admin"
|
||||
|
||||
# Alice cannot create a token for admin
|
||||
_, res = await client.post(
|
||||
"/auth/tokens",
|
||||
json={"username": "admin", "name": "impersonation"},
|
||||
headers={"Authorization": _basic_auth("alice", "secret")},
|
||||
)
|
||||
assert res.status_code == 403
|
||||
|
||||
# Alice cannot delete admin's token
|
||||
_, res = await client.delete(
|
||||
f"/auth/tokens/{admin_token_id}",
|
||||
headers={"Authorization": _basic_auth("alice", "secret")},
|
||||
)
|
||||
assert res.status_code == 403
|
||||
|
||||
# Alice can delete her own token
|
||||
_, res = await client.delete(
|
||||
f"/auth/tokens/{alice_token_id}",
|
||||
headers={"Authorization": _basic_auth("alice", "secret")},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
|
||||
# Alice's token auth still works until deletion is processed
|
||||
# Verify token auth worked during the test
|
||||
_, res = await client.get(
|
||||
"/auth/tokens",
|
||||
headers={"Authorization": _basic_auth("token", alice_token_key)},
|
||||
)
|
||||
# Token was deleted above, so this should now be unauthenticated
|
||||
# Actually the token key lookup will fail, and since there's no session fallback...
|
||||
# With auth header present but invalid, it should return 401
|
||||
assert res.status_code == 401
|
||||
Reference in New Issue
Block a user