# /// script
# requires-python = ">=3.14"
# dependencies = ["falcon>=4.0", "uvicorn[standard]", "typer", "argon2-cffi", "lxml"]
# ///
"""RFC 4918 compliant WebDAV server using Falcon ASGI."""

from __future__ import annotations

import asyncio
import base64
import collections
import concurrent.futures
import errno
import hashlib
import hmac
import datetime
import html as _html
import logging
import mimetypes
import os
import pathlib
import secrets
import stat
import re
import shutil
import tempfile
import threading
import time
import urllib.parse
import uuid
from lxml import etree
from lxml.builder import ElementMaker
from dataclasses import dataclass, field
from email.utils import formatdate

import falcon
import argon2
import argon2.exceptions
import falcon.asgi
import typer
import uvicorn

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

D = "{DAV:}"
E = ElementMaker(namespace="DAV:", nsmap={"D": "DAV:"})
_SECURE_PARSER = etree.XMLParser(resolve_entities=False, no_network=True)

WRITE_METHODS = frozenset(
    {"PUT", "DELETE", "MKCOL", "COPY", "MOVE", "PROPPATCH", "LOCK", "UNLOCK"}
)
ALL_METHODS = (
    "GET, HEAD, OPTIONS, PUT, DELETE, MKCOL, COPY, MOVE, "
    "PROPFIND, PROPPATCH, LOCK, UNLOCK"
)
LOCK_DEFAULT_TIMEOUT = 600  # seconds
LOCK_MAX_TIMEOUT = 3600  # 1 hour ceiling — bounds lock-table fill time
MAX_UPLOAD_BYTES = 8 * 1024 * 1024 * 1024  # 8 GiB hard ceiling per PUT
MAX_XML_BODY_BYTES = 1_048_576  # 1 MiB ceiling for PROPFIND/PROPPATCH/LOCK/MKCOL bodies
MAX_LOCKS = 2048  # per-server lock table ceiling
MAX_LOCKS_PER_IP = 32  # per-client active-lock cap
MAX_PROPS_PER_RESOURCE = 256  # dead property ceiling per resource
MAX_PROP_VALUE_BYTES = 65_536  # 64 KiB per property value
MAX_PROP_CHILDREN = 1024  # element ceiling inside <prop> tree
STREAM_CHUNK = 64 * 1024  # 64 KiB chunk for streaming I/O
AUTH_FAIL_WINDOW = 60.0  # seconds for sliding failed-auth window
AUTH_FAIL_MAX = 10  # max failed-auth attempts in window per IP
AUTH_CACHE_MAX = 4  # verified-credential cache cap (one entry per principal)
AUTH_CACHE_TTL = 300.0  # seconds an Authorization header stays cached
TMP_DIRNAME = ".dav-tmp"  # hidden in-tree temp dir for atomic ops


def _unlink_quiet(path: pathlib.Path) -> None:
    try:
        os.unlink(str(path))
    except OSError:
        pass


def _rmtree_quiet(path: pathlib.Path) -> None:
    try:
        shutil.rmtree(str(path))
    except OSError:
        pass


async def _file_stream(path: pathlib.Path, start: int, length: int):
    """Async byte iterator over a file region. Yields STREAM_CHUNK at a time."""
    fd = await asyncio.to_thread(os.open, str(path), os.O_RDONLY)
    try:
        if start:
            await asyncio.to_thread(os.lseek, fd, start, os.SEEK_SET)
        remaining = length
        while remaining > 0:
            n = STREAM_CHUNK if remaining > STREAM_CHUNK else remaining
            chunk = await asyncio.to_thread(os.read, fd, n)
            if not chunk:
                break
            remaining -= len(chunk)
            yield chunk
    finally:
        await asyncio.to_thread(os.close, fd)


def _atomic_touch(path: pathlib.Path) -> None:
    """Atomically create an empty file. Raises FileExistsError if the file
    already exists. Uses O_CREAT|O_EXCL so concurrent calls cannot race."""
    fd = os.open(str(path), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
    os.close(fd)


async def _read_xml_body(req: falcon.asgi.Request) -> bytes:
    """Read up to MAX_XML_BODY_BYTES from the request body. Raise 413 if oversized."""
    body = await req.bounded_stream.read(MAX_XML_BODY_BYTES + 1)
    if len(body) > MAX_XML_BODY_BYTES:
        raise falcon.HTTPPayloadTooLarge(
            description=f"Request body exceeds {MAX_XML_BODY_BYTES} bytes"
        )
    return body


def _world_visible(st: os.stat_result) -> bool:
    if stat.S_ISDIR(st.st_mode):
        return (st.st_mode & 0o005) == 0o005
    return bool(st.st_mode & 0o004)


# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------


@dataclass
class LockInfo:
    token: str
    path: str
    owner_xml: str
    depth: str
    scope: str  # "exclusive" or "shared"
    timeout: float  # absolute time.time() when lock expires
    principal: str | None = None  # authenticated user that holds the lock
    remote_addr: str | None = None  # IP that obtained the lock (for per-IP cap)


# ---------------------------------------------------------------------------
# LockManager
# ---------------------------------------------------------------------------


class LockManager:
    def __init__(self) -> None:
        self._locks: dict[str, LockInfo] = {}  # token -> LockInfo
        self._locks_by_ip: dict[str, int] = {}
        self._mu = threading.Lock()

    def _expire_unlocked(self) -> None:
        now = time.time()
        expired = [t for t, lk in self._locks.items() if lk.timeout <= now]
        for t in expired:
            lk = self._locks.pop(t)
            if lk.remote_addr:
                self._dec_ip_unlocked(lk.remote_addr)

    def _dec_ip_unlocked(self, ip: str) -> None:
        n = self._locks_by_ip.get(ip, 0) - 1
        if n <= 0:
            self._locks_by_ip.pop(ip, None)
        else:
            self._locks_by_ip[ip] = n

    def cleanup(self) -> None:
        with self._mu:
            self._expire_unlocked()

    def acquire(
        self,
        path: str,
        owner_xml: str,
        depth: str,
        scope: str,
        timeout_secs: int = LOCK_DEFAULT_TIMEOUT,
        principal: str | None = None,
        remote_addr: str | None = None,
    ) -> LockInfo | None:
        with self._mu:
            self._expire_unlocked()
            if len(self._locks) >= MAX_LOCKS:
                return None
            if remote_addr and self._locks_by_ip.get(remote_addr, 0) >= MAX_LOCKS_PER_IP:
                return None
            for lk in self._locks.values():
                if self._conflicts(path, depth, lk):
                    if scope == "exclusive" or lk.scope == "exclusive":
                        return None
            token = f"urn:uuid:{uuid.uuid4()}"
            lock = LockInfo(
                token=token,
                path=path,
                owner_xml=owner_xml,
                depth=depth,
                scope=scope,
                timeout=time.time() + timeout_secs,
                principal=principal,
                remote_addr=remote_addr,
            )
            self._locks[token] = lock
            if remote_addr:
                self._locks_by_ip[remote_addr] = self._locks_by_ip.get(remote_addr, 0) + 1
            return lock

    def refresh(
        self,
        token: str,
        principal: str | None,
        timeout_secs: int = LOCK_DEFAULT_TIMEOUT,
    ) -> LockInfo | None:
        with self._mu:
            self._expire_unlocked()
            lock = self._locks.get(token)
            if lock is None:
                return None
            if lock.principal != principal:
                return None
            lock.timeout = time.time() + timeout_secs
            return lock

    def release(self, token: str, principal: str | None) -> str:
        """Return 'ok', 'forbidden' (principal mismatch), or 'missing'."""
        with self._mu:
            lock = self._locks.get(token)
            if lock is None:
                return "missing"
            if lock.principal != principal:
                return "forbidden"
            del self._locks[token]
            if lock.remote_addr:
                self._dec_ip_unlocked(lock.remote_addr)
            return "ok"

    def get_locks(self, path: str) -> list[LockInfo]:
        with self._mu:
            self._expire_unlocked()
            result = []
            for lk in self._locks.values():
                if lk.path == path:
                    result.append(lk)
                elif lk.depth == "infinity" and path.startswith(lk.path.rstrip("/") + "/"):
                    result.append(lk)
            return result

    def check_locked(self, path: str, if_header: str | None) -> bool:
        """Return True if the operation is allowed (lock-wise).

        RFC 4918 §7.4: any lock on the path (exclusive or shared) requires
        the caller to present a matching token in the If header.
        """
        locks = self.get_locks(path)
        if not locks:
            return True
        if not if_header:
            return False
        tokens = set(re.findall(r"<([^>]+)>", if_header))
        for lk in locks:
            if lk.token not in tokens:
                return False
        return True

    def remove_for_path(self, path: str) -> None:
        with self._mu:
            prefix = path.rstrip("/") + "/"
            to_del = [
                t
                for t, lk in self._locks.items()
                if lk.path == path or lk.path.startswith(prefix)
            ]
            for t in to_del:
                lk = self._locks.pop(t)
                if lk.remote_addr:
                    self._dec_ip_unlocked(lk.remote_addr)

    def move_locks(self, src: str, dst: str) -> None:
        with self._mu:
            src_prefix = src.rstrip("/") + "/"
            updates: list[tuple[str, str]] = []
            for lk in self._locks.values():
                if lk.path == src:
                    updates.append((lk.token, dst))
                elif lk.path.startswith(src_prefix):
                    new_path = dst.rstrip("/") + "/" + lk.path[len(src_prefix) :]
                    updates.append((lk.token, new_path))
            for token, new_path in updates:
                self._locks[token].path = new_path

    @staticmethod
    def _conflicts(path: str, depth: str, existing: LockInfo) -> bool:
        ep = existing.path
        if ep == path:
            return True
        if existing.depth == "infinity" and path.startswith(ep.rstrip("/") + "/"):
            return True
        if depth == "infinity" and ep.startswith(path.rstrip("/") + "/"):
            return True
        return False


# ---------------------------------------------------------------------------
# PropertyManager (dead properties, in-memory only)
# ---------------------------------------------------------------------------


@dataclass
class PropertyManager:
    _props: dict[str, dict[str, str]] = field(default_factory=dict)

    def set_prop(self, path: str, key: str, value: str) -> None:
        bucket = self._props.setdefault(path, {})
        if len(value) > MAX_PROP_VALUE_BYTES:
            raise falcon.HTTPPayloadTooLarge(
                description="Property value exceeds size limit"
            )
        if key not in bucket and len(bucket) >= MAX_PROPS_PER_RESOURCE:
            raise falcon.HTTPInsufficientStorage(
                description="Too many properties on this resource"
            )
        bucket[key] = value

    def remove_prop(self, path: str, key: str) -> bool:
        if path in self._props and key in self._props[path]:
            del self._props[path][key]
            return True
        return False

    def get_all(self, path: str) -> dict[str, str]:
        return dict(self._props.get(path, {}))

    def delete(self, path: str) -> None:
        prefix = path.rstrip("/") + "/"
        to_del = [p for p in self._props if p == path or p.startswith(prefix)]
        for p in to_del:
            del self._props[p]

    def move(self, src: str, dst: str) -> None:
        src_prefix = src.rstrip("/") + "/"
        moves: list[tuple[str, str]] = []
        for p in list(self._props):
            if p == src:
                moves.append((p, dst))
            elif p.startswith(src_prefix):
                moves.append((p, dst.rstrip("/") + "/" + p[len(src_prefix) :]))
        for old, new in moves:
            self._props[new] = self._props.pop(old)

    def copy(self, src: str, dst: str) -> None:
        """Duplicate dead properties from src tree to dst tree."""
        src_prefix = src.rstrip("/") + "/"
        copies: list[tuple[str, str]] = []
        for p in list(self._props):
            if p == src:
                copies.append((p, dst))
            elif p.startswith(src_prefix):
                copies.append((p, dst.rstrip("/") + "/" + p[len(src_prefix) :]))
        for old, new in copies:
            self._props[new] = dict(self._props[old])


# ---------------------------------------------------------------------------
# XML helpers
# ---------------------------------------------------------------------------


def _content_type(fspath: pathlib.Path) -> str:
    if fspath.is_dir():
        return "httpd/unix-directory"
    mt = mimetypes.guess_type(fspath.name)[0] or "application/octet-stream"
    # Tag text payloads with UTF-8. Without it, browsers and many clients
    # fall back to latin-1 / windows-1252 and any non-ASCII byte renders
    # as mojibake (e.g. "café" → "café").
    if mt.startswith("text/") or mt in (
        "application/json",
        "application/xml",
        "application/javascript",
        "application/x-yaml",
    ):
        return f"{mt}; charset=utf-8"
    return mt


def _etag(st: os.stat_result, secret: bytes) -> str:
    """HMAC-keyed ETag. Stable for a server lifetime; opaque to clients —
    inode + mtime + size are no longer recoverable from the digest."""
    raw = f"{st.st_ino}-{st.st_mtime_ns}-{st.st_size}".encode()
    return f'"{hmac.new(secret, raw, hashlib.sha256).hexdigest()[:32]}"'


def _http_date(ts: float) -> str:
    return formatdate(ts, usegmt=True)


def _iso_date(ts: float) -> str:
    return datetime.datetime.fromtimestamp(ts, datetime.UTC).strftime(
        "%Y-%m-%dT%H:%M:%SZ"
    )


def _supported_lock_element() -> etree._Element:
    return E.supportedlock(
        *[
            E.lockentry(
                E.lockscope(getattr(E, scope_name)()),
                E.locktype(E.write()),
            )
            for scope_name in ("exclusive", "shared")
        ]
    )


def _lock_discovery_element(
    locks: list[LockInfo], viewer_principal: str | None = None
) -> etree._Element:
    """Build a <D:lockdiscovery> element. Token element is redacted unless
    the viewer is the principal that holds the lock."""
    ld = E.lockdiscovery()
    for lk in locks:
        owner_el = E.owner()
        if lk.owner_xml:
            try:
                owner_el.append(etree.fromstring(lk.owner_xml, parser=_SECURE_PARSER))
            except etree.XMLSyntaxError:
                owner_el.text = lk.owner_xml
        remaining = max(0, int(lk.timeout - time.time()))
        children: list[etree._Element] = [
            E.locktype(E.write()),
            E.lockscope(getattr(E, lk.scope)()),
            E.depth(lk.depth),
            owner_el,
            E.timeout(f"Second-{remaining}"),
        ]
        # Only the owning principal may see the token. Everyone else gets the
        # rest of the lockdiscovery, just without the bearer credential.
        if viewer_principal is not None and viewer_principal == lk.principal:
            children.append(E.locktoken(E.href(lk.token)))
        children.append(E.lockroot(E.href(lk.path)))
        ld.append(E.activelock(*children))
    return ld


def _build_prop_response(
    href: str,
    props_ok: dict[str, etree._Element | str | None],
    props_404: list[str],
) -> etree._Element:
    response = E.response(E.href(href))

    if props_ok:
        prop = E.prop()
        for name, val in props_ok.items():
            if isinstance(val, etree._Element):
                prop.append(val)
            elif val is not None:
                child = etree.SubElement(prop, f"{D}{name}")
                child.text = str(val)
            else:
                etree.SubElement(prop, f"{D}{name}")
        response.append(E.propstat(prop, E.status("HTTP/1.1 200 OK")))

    if props_404:
        prop = E.prop()
        for name in props_404:
            etree.SubElement(prop, f"{D}{name}")
        response.append(E.propstat(prop, E.status("HTTP/1.1 404 Not Found")))

    return response


def _build_multistatus(responses: list[etree._Element]) -> bytes:
    return etree.tostring(
        E.multistatus(*responses),
        xml_declaration=True,
        encoding="utf-8",
    )


def _parse_propfind(body: bytes) -> tuple[str, set[str] | None]:
    """Return (mode, requested_props).

    mode is one of "allprop", "propname", or "prop".
    For allprop/propname, requested_props is None.
    For prop, requested_props is a set of local tag names.
    """
    if not body or not body.strip():
        return "allprop", None
    try:
        root = etree.fromstring(body, parser=_SECURE_PARSER)
    except etree.XMLSyntaxError:
        return "allprop", None

    if root.find(f"{D}allprop") is not None:
        return "allprop", None
    if root.find(f"{D}propname") is not None:
        return "propname", None
    prop_el = root.find(f"{D}prop")
    if prop_el is not None:
        names = set()
        for i, child in enumerate(prop_el):
            if i >= MAX_PROP_CHILDREN:
                raise falcon.HTTPBadRequest(
                    description="Too many properties requested"
                )
            tag = child.tag
            if tag.startswith(f"{D}"):
                names.add(tag[len(D) :])
            else:
                names.add(tag)
        return "prop", names
    return "allprop", None


def _parse_timeout(header: str | None) -> int:
    if not header:
        return LOCK_DEFAULT_TIMEOUT
    for part in header.split(","):
        part = part.strip()
        if part.lower().startswith("second-"):
            try:
                requested = int(part[7:])
            except ValueError:
                continue
            return max(1, min(requested, LOCK_MAX_TIMEOUT))
    return LOCK_DEFAULT_TIMEOUT


def _parse_lockinfo(body: bytes) -> tuple[str, str]:
    """Return (scope, owner_xml)."""
    scope = "exclusive"
    owner_xml = ""
    if not body or not body.strip():
        return scope, owner_xml
    try:
        root = etree.fromstring(body, parser=_SECURE_PARSER)
    except etree.XMLSyntaxError:
        return scope, owner_xml
    scope_el = root.find(f"{D}lockscope")
    if scope_el is not None:
        if scope_el.find(f"{D}shared") is not None:
            scope = "shared"
    owner_el = root.find(f"{D}owner")
    if owner_el is not None:
        children = list(owner_el)
        if children:
            owner_xml = etree.tostring(children[0], encoding="unicode")
        elif owner_el.text:
            owner_xml = owner_el.text
    return scope, owner_xml


def _info_page(
    req: falcon.asgi.Request,
    mode: str,
    basedir: pathlib.Path,
) -> tuple[str, str]:
    """Return (content_type, body) for a human/agent-facing info response.

    Negotiates on Accept: text/html → rendered HTML, else plain markdown.
    """
    host = req.get_header("Host") or "localhost"
    scheme = (
        "https"
        if req.forwarded_prefix and req.forwarded_prefix.startswith("https")
        else "http"
    )
    base_url = f"{scheme}://{host}/"

    md = f"""\
# WebDAV Server

**Access mode:** {mode}
**Base URL:** {base_url}

## Connecting

### macOS
Finder: Go → Connect to Server → `{base_url}`

### Linux
```sh
# Mount (davfs2)
sudo apt install davfs2
sudo mkdir /mnt/webdav
sudo mount -t davfs {base_url} /mnt/webdav

# GUI: Nautilus address bar
davs://{host}/
# GUI: Dolphin address bar
webdavs://{host}/

# CLI
cadaver {base_url}
```

### Windows
```
# Map Network Drive → Folder:
{base_url}

# Or via command line:
net use Z: {base_url}
```
Note: if Windows refuses to connect, set registry key
`HKLM\\SYSTEM\\CurrentControlSet\\Services\\WebClient\\Parameters\\BasicAuthLevel = 2`
and restart the WebClient service.

### Cross-platform (rclone)
```sh
rclone copy {base_url} ./local-copy --webdav-vendor other
```
Cyberduck and WinSCP also support WebDAV via the GUI.

### Agents / programmatic access
```sh
# List directory
curl -X PROPFIND -H "Depth: 1" {base_url}

# Download a file
curl {base_url}path/to/file

# Upload a file (requires auth if server has credentials)
curl -T localfile {base_url}path/to/file

# Create a directory
curl -X MKCOL {base_url}newdir/
```

## Source Code

[Download webdav.py]({base_url}webdav.py) — self-contained, PEP 723, runs with `uv run webdav.py`.

## Supported Methods

`GET` `HEAD` `PUT` `DELETE` `MKCOL` `COPY` `MOVE`
`PROPFIND` `PROPPATCH` `LOCK` `UNLOCK` `OPTIONS`

## Protocol

RFC 4918 WebDAV — DAV compliance classes 1, 2, 3.
Use `PROPFIND` with `Depth: 1` to list collections.
Use `PROPFIND` with `Depth: 0` to inspect a single resource.
"""

    accept = req.get_header("Accept") or ""
    if "text/html" in accept:
        # Escape all user-controlled values before embedding in HTML.
        h_url = _html.escape(base_url, quote=True)
        h_host = _html.escape(host, quote=True)
        h_mode = _html.escape(mode)
        page = f"""\
<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>WebDAV Server</title>
  <style>
    body {{ font-family: system-ui, sans-serif; max-width: 640px; margin: 2rem auto; padding: 0 1rem; color: #1a1a1a; }}
    h1 {{ border-bottom: 2px solid #e0e0e0; padding-bottom: .5rem; }}
    h2 {{ margin-top: 1.5rem; }}
    h3 {{ margin-top: 1rem; margin-bottom: .25rem; }}
    code {{ background: #f4f4f4; padding: .1em .4em; border-radius: 3px; font-size: .9em; }}
    pre {{ background: #f4f4f4; padding: 1rem; border-radius: 4px; overflow-x: auto; }}
    table {{ border-collapse: collapse; width: 100%; margin-bottom: 1rem; }}
    td, th {{ padding: .4rem .8rem; border: 1px solid #ddd; text-align: left; }}
    th {{ background: #f8f8f8; }}
    .note {{ color: #666; font-size: .9em; }}
  </style>
</head>
<body>
  <h1>WebDAV Server</h1>
  <table>
    <tr><th>Access mode</th><td>{h_mode}</td></tr>
    <tr><th>Base URL</th><td><a href="{h_url}"><code>{h_url}</code></a></td></tr>
  </table>

  <h2>Connecting</h2>

  <h3>macOS</h3>
  <p>Finder: <strong>Go → Connect to Server</strong> → <code>{h_url}</code></p>

  <h3>Linux</h3>
  <pre># Mount (davfs2)
sudo apt install davfs2
sudo mkdir /mnt/webdav
sudo mount -t davfs {h_url} /mnt/webdav

# GUI: Nautilus address bar → davs://{h_host}/
# GUI: Dolphin address bar  → webdavs://{h_host}/

# CLI
cadaver {h_url}</pre>

  <h3>Windows</h3>
  <p>Map Network Drive → Folder: <code>{h_url}</code><br>
  Or: <code>net use Z: {h_url}</code></p>
  <p class="note">If Windows refuses to connect: set
  <code>HKLM\\SYSTEM\\CurrentControlSet\\Services\\WebClient\\Parameters\\BasicAuthLevel = 2</code>
  and restart the WebClient service.</p>

  <h3>Cross-platform (rclone)</h3>
  <pre>rclone copy {h_url} ./local-copy --webdav-vendor other</pre>
  <p class="note">Cyberduck and WinSCP also support WebDAV via GUI.</p>

  <h3>Agents / programmatic access</h3>
  <pre># List directory
curl -X PROPFIND -H "Depth: 1" {h_url}

# Download a file
curl {h_url}path/to/file

# Upload (requires auth if credentials are configured)
curl -T localfile {h_url}path/to/file

# Create a directory
curl -X MKCOL {h_url}newdir/</pre>

  <h2>Source Code</h2>
  <p><a href="{h_url}webdav.py">Download webdav.py</a> — self-contained, PEP 723, runs with <code>uv run webdav.py</code>.</p>

  <h2>Supported Methods</h2>
  <pre>GET  HEAD  PUT  DELETE  MKCOL  COPY  MOVE
PROPFIND  PROPPATCH  LOCK  UNLOCK  OPTIONS</pre>
  <p>RFC 4918 WebDAV — DAV compliance classes 1, 2, 3.</p>
</body>
</html>
"""
        return "text/html; charset=utf-8", page

    return "text/markdown; charset=utf-8", md


def _lock_response_body(lock: LockInfo) -> bytes:
    # The LOCK response always shows the token to the principal that just
    # obtained or refreshed it.
    return etree.tostring(
        E.prop(_lock_discovery_element([lock], viewer_principal=lock.principal)),
        xml_declaration=True,
        encoding="utf-8",
    )


# ---------------------------------------------------------------------------
# Middleware
# ---------------------------------------------------------------------------


class AuthMiddleware:
    """Unified auth middleware covering all four access modes.

    | credentials | allow_anonymous_read | reads      | writes          |
    |-------------|----------------------|------------|-----------------|
    | set         | False                | auth req'd | auth req'd      |
    | set         | True                 | open       | auth req'd      |
    | not set     | True                 | open       | 403 (no creds)  |
    | not set     | False                | —          | refused at CLI  |
    """

    _ph = argon2.PasswordHasher()
    # Bounded executor for argon2.verify so an auth flood can't saturate the
    # default thread pool that file I/O also uses.
    _argon_pool = concurrent.futures.ThreadPoolExecutor(
        max_workers=4, thread_name_prefix="argon2"
    )

    def __init__(
        self,
        username: str | None,
        password: str | None,
        allow_anonymous_read: bool,
    ) -> None:
        self._has_creds = username is not None
        self._anon_read = allow_anonymous_read
        # Bind username into the secret so a single verify call checks both.
        # Neither username nor password is kept in memory after this point.
        if username is not None and password is not None:
            self._hash: str | None = self._ph.hash(f"{username}:{password}")
        else:
            self._hash = None
        # Sliding-window per-IP failed-auth tracker.
        self._fail_log: dict[str, collections.deque[float]] = {}
        self._fail_mu = threading.Lock()
        # Verified-credential cache. Key is sha256 of the Authorization header
        # value; presence implies the header was already accepted by argon2
        # within the TTL window. Failures are never cached, so brute force
        # still pays full KDF cost and feeds the per-IP fail counter.
        self._auth_cache: dict[bytes, tuple[str, float]] = {}
        self._cache_mu = threading.Lock()

    def _client_ip(self, req: falcon.asgi.Request) -> str:
        route = req.access_route
        if route:
            return route[-1]
        return req.remote_addr or "?"

    def _record_fail(self, ip: str) -> int:
        now = time.time()
        with self._fail_mu:
            dq = self._fail_log.setdefault(ip, collections.deque())
            cutoff = now - AUTH_FAIL_WINDOW
            while dq and dq[0] < cutoff:
                dq.popleft()
            dq.append(now)
            return len(dq)

    def _is_blocked(self, ip: str) -> bool:
        now = time.time()
        with self._fail_mu:
            dq = self._fail_log.get(ip)
            if not dq:
                return False
            cutoff = now - AUTH_FAIL_WINDOW
            while dq and dq[0] < cutoff:
                dq.popleft()
            return len(dq) >= AUTH_FAIL_MAX

    def _clear_fails(self, ip: str) -> None:
        with self._fail_mu:
            self._fail_log.pop(ip, None)

    async def process_request(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response
    ) -> None:
        # Default: no authenticated principal. Set on every exit path so
        # downstream code (locks, error handlers) can rely on this attribute.
        req.context.user = None

        is_write = req.method in WRITE_METHODS

        # Writes with no credentials configured → always forbidden
        if is_write and not self._has_creds:
            raise falcon.HTTPForbidden(
                description="Server has no write credentials configured"
            )

        auth = req.get_header("Authorization")
        has_auth_header = bool(auth and auth.startswith("Basic "))

        # Reads with anonymous access enabled and no credentials presented:
        # let the request through unauthenticated.
        if not is_write and self._anon_read and not has_auth_header:
            return

        ip = self._client_ip(req)
        if self._is_blocked(ip):
            resp.set_header("Retry-After", str(int(AUTH_FAIL_WINDOW)))
            raise falcon.HTTPTooManyRequests(
                description="Too many failed authentication attempts"
            )

        # All remaining requests — and any request that presented an
        # Authorization header — require valid credentials so the principal
        # binding is trustworthy.
        if not has_auth_header:
            resp.set_header("WWW-Authenticate", 'Basic realm="WebDAV"')
            raise falcon.HTTPUnauthorized(description="Authentication required")

        # Fast path: this exact Authorization header was already verified by
        # argon2 within the TTL window. Constant-time key compare via dict
        # hash → returned principal is bound to the cached entry.
        cache_key = hashlib.sha256(auth.encode("utf-8")).digest()
        now = time.time()
        with self._cache_mu:
            hit = self._auth_cache.get(cache_key)
            if hit is not None and hit[1] > now:
                req.context.user = hit[0]
                return
            elif hit is not None:
                # Expired entry — drop it eagerly.
                self._auth_cache.pop(cache_key, None)

        try:
            decoded = base64.b64decode(auth[6:]).decode("utf-8", errors="replace")
        except Exception:
            self._record_fail(ip)
            resp.set_header("WWW-Authenticate", 'Basic realm="WebDAV"')
            raise falcon.HTTPUnauthorized(description="Invalid credentials")
        supplied_user, _, supplied_pw = decoded.partition(":")
        # Single verify call covers both username and password. Run on a
        # bounded executor — Argon2 is intentionally CPU-intensive and would
        # otherwise saturate the default thread pool that file I/O uses too.
        loop = asyncio.get_running_loop()
        try:
            await loop.run_in_executor(
                self._argon_pool,
                self._ph.verify,
                self._hash,
                f"{supplied_user}:{supplied_pw}",
            )
        except (
            argon2.exceptions.VerifyMismatchError,
            argon2.exceptions.VerificationError,
        ):
            self._record_fail(ip)
            resp.set_header("WWW-Authenticate", 'Basic realm="WebDAV"')
            raise falcon.HTTPUnauthorized(description="Invalid credentials")
        # Successful auth — reset the IP's failure window, bind principal,
        # and remember this header so subsequent requests skip argon2.
        self._clear_fails(ip)
        req.context.user = supplied_user
        with self._cache_mu:
            self._auth_cache[cache_key] = (supplied_user, now + AUTH_CACHE_TTL)
            if len(self._auth_cache) > AUTH_CACHE_MAX:
                expired = [
                    k for k, (_, exp) in self._auth_cache.items() if exp <= now
                ]
                if expired:
                    for k in expired:
                        self._auth_cache.pop(k, None)
                else:
                    oldest = min(
                        self._auth_cache, key=lambda k: self._auth_cache[k][1]
                    )
                    self._auth_cache.pop(oldest, None)


# ---------------------------------------------------------------------------
# WebDAV resource
# ---------------------------------------------------------------------------


class WebDAVResource:
    def __init__(
        self,
        basedir: pathlib.Path,
        lock_mgr: LockManager,
        prop_mgr: PropertyManager,
        mode: str = "unknown",
        hide_dot_paths: bool = True,
    ) -> None:
        self.basedir = basedir
        self.lock_mgr = lock_mgr
        self.prop_mgr = prop_mgr
        self.mode = mode
        self.hide_dot_paths = hide_dot_paths
        self.tmp_dir = basedir / TMP_DIRNAME
        self._etag_secret = secrets.token_bytes(32)

    def etag(self, st: os.stat_result) -> str:
        return _etag(st, self._etag_secret)

    # -- helpers -----------------------------------------------------------

    def _resolve(self, path: str) -> pathlib.Path:
        decoded = urllib.parse.unquote(path)
        resolved = (self.basedir / decoded).resolve()
        if not resolved.is_relative_to(self.basedir):
            raise falcon.HTTPForbidden(description="Path traversal denied")
        # The hidden temp dir is internal scaffolding; never expose it to
        # WebDAV callers regardless of hide_dot_paths.
        try:
            rel = resolved.relative_to(self.basedir)
        except ValueError:
            raise falcon.HTTPForbidden(description="Path traversal denied")
        if rel.parts and rel.parts[0] == TMP_DIRNAME:
            raise falcon.HTTPNotFound()
        cur = resolved
        while cur != self.basedir:
            try:
                st = cur.stat()
            except FileNotFoundError:
                break
            except OSError:
                raise falcon.HTTPNotFound()
            if (self.hide_dot_paths and cur.name.startswith(".")) or not _world_visible(st):
                raise falcon.HTTPNotFound()
            cur = cur.parent
        return resolved

    def _relpath(self, fspath: pathlib.Path) -> str:
        rel = fspath.relative_to(self.basedir).as_posix()
        return "/" if rel == "." else "/" + rel

    def _href(self, fspath: pathlib.Path) -> str:
        p = urllib.parse.quote(self._relpath(fspath), safe="/")
        if fspath.is_dir() and not p.endswith("/"):
            p += "/"
        return p

    def _dav_path(self, fspath: pathlib.Path) -> str:
        return self._relpath(fspath)

    def _get_props(
        self,
        fspath: pathlib.Path,
        st: os.stat_result,
        mode: str,
        requested: set[str] | None,
        viewer_principal: str | None = None,
    ) -> tuple[dict[str, etree._Element | str | None], list[str]]:
        is_dir = fspath.is_dir()
        etag = self.etag(st)

        live: dict[str, etree._Element | str | None] = {}
        live["resourcetype"] = E.resourcetype(E.collection()) if is_dir else E.resourcetype()
        live["getlastmodified"] = _http_date(st.st_mtime)
        live["creationdate"] = _iso_date(st.st_ctime)
        live["displayname"] = fspath.name or "/"
        live["getetag"] = etag
        live["getcontenttype"] = _content_type(fspath)
        if not is_dir:
            live["getcontentlength"] = str(st.st_size)
        live["getcontentlanguage"] = None  # not set
        live["supportedlock"] = _supported_lock_element()
        locks = self.lock_mgr.get_locks(self._dav_path(fspath))
        live["lockdiscovery"] = _lock_discovery_element(locks, viewer_principal)

        # Dead properties
        dead = self.prop_mgr.get_all(self._dav_path(fspath))

        if mode == "propname":
            return {k: None for k in (*live, *dead)}, []

        if mode == "allprop":
            return {**live, **dead}, []

        # mode == "prop"
        props_ok: dict[str, etree._Element | str | None] = {}
        props_404: list[str] = []
        if requested:
            for name in requested:
                if name in live:
                    props_ok[name] = live[name]
                elif name in dead:
                    props_ok[name] = dead[name]
                else:
                    props_404.append(name)
        return props_ok, props_404

    def _parse_destination(self, req: falcon.asgi.Request) -> pathlib.Path:
        dest = req.get_header("Destination")
        if not dest:
            raise falcon.HTTPBadRequest(description="Missing Destination header")
        return self._resolve(urllib.parse.urlparse(dest).path.lstrip("/"))

    @staticmethod
    def _set_lock_response(
        resp: falcon.asgi.Response, lock: LockInfo, status: str
    ) -> None:
        resp.status = status
        resp.content_type = "application/xml; charset=utf-8"
        resp.set_header("Lock-Token", f"<{lock.token}>")
        resp.data = _lock_response_body(lock)

    def _check_lock(self, req: falcon.asgi.Request, fspath: pathlib.Path) -> None:
        dav_path = self._dav_path(fspath)
        if_header = req.get_header("If")
        if not self.lock_mgr.check_locked(dav_path, if_header):
            raise falcon.HTTPLocked()

    # -- OPTIONS -----------------------------------------------------------

    async def on_options(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        resp.status = falcon.HTTP_200
        resp.set_header("DAV", "1, 2, 3")
        resp.set_header("Allow", ALL_METHODS)
        resp.set_header("MS-Author-Via", "DAV")
        resp.content_length = 0

    # -- HEAD --------------------------------------------------------------

    # -- HEAD / GET --------------------------------------------------------

    async def on_head(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        fspath = self._resolve(kwargs.get("path", ""))
        if not fspath.exists():
            raise falcon.HTTPNotFound()
        if fspath.is_dir():
            # Mirror GET on a collection: return the info-page metadata
            # without the body.
            ctype, body = _info_page(req, self.mode, self.basedir)
            resp.set_header("Vary", "Accept")
            resp.content_type = ctype
            resp.content_length = len(body.encode("utf-8"))
            return
        st = await asyncio.to_thread(os.stat, fspath)
        resp.set_header("ETag", self.etag(st))
        resp.set_header("Last-Modified", _http_date(st.st_mtime))
        resp.set_header("Accept-Ranges", "bytes")
        resp.content_type = _content_type(fspath)
        resp.content_length = st.st_size

    async def on_get(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        fspath = self._resolve(kwargs.get("path", ""))
        if not fspath.exists():
            raise falcon.HTTPNotFound()
        if fspath.is_dir():
            ctype, body = _info_page(req, self.mode, self.basedir)
            resp.set_header("Vary", "Accept")
            resp.content_type = ctype
            resp.text = body
            return

        st = await asyncio.to_thread(os.stat, fspath)
        etag = self.etag(st)
        resp.set_header("ETag", etag)
        resp.set_header("Last-Modified", _http_date(st.st_mtime))
        resp.set_header("Accept-Ranges", "bytes")
        resp.content_type = _content_type(fspath)

        # Conditional request support
        inm = req.get_header("If-None-Match")
        if inm and etag in [t.strip() for t in inm.split(",")]:
            resp.status = falcon.HTTP_304
            return

        total = st.st_size
        start = 0
        end = total - 1 if total > 0 else 0
        partial = False

        range_header = req.get_header("Range")
        if range_header and range_header.startswith("bytes="):
            try:
                range_spec = range_header[6:]
                start_s, _, end_s = range_spec.partition("-")
                if start_s:
                    rs = int(start_s)
                    re_ = int(end_s) if end_s else total - 1
                else:
                    suffix = int(end_s)
                    rs = max(0, total - suffix)
                    re_ = total - 1
            except ValueError:
                resp.status = "416 Range Not Satisfiable"
                resp.set_header("Content-Range", f"bytes */{total}")
                return
            if rs < 0 or rs >= total or re_ >= total or rs > re_:
                resp.status = "416 Range Not Satisfiable"
                resp.set_header("Content-Range", f"bytes */{total}")
                return
            start, end = rs, re_
            partial = True
            resp.status = "206 Partial Content"
            resp.set_header("Content-Range", f"bytes {start}-{end}/{total}")

        length = end - start + 1 if total > 0 else 0
        resp.content_length = length
        if length == 0:
            resp.data = b""
            return
        resp.stream = _file_stream(fspath, start, length)

    # -- PUT ---------------------------------------------------------------

    async def on_put(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        fspath = self._resolve(kwargs.get("path", ""))
        self._check_lock(req, fspath)
        parent = fspath.parent
        if not parent.exists():
            raise falcon.HTTPConflict(description="Parent collection does not exist")
        if fspath.is_dir():
            raise falcon.HTTPConflict(description="Cannot PUT to a collection")

        # Conditional request (initial check)
        im = req.get_header("If-Match")
        if im:
            if fspath.exists():
                st = await asyncio.to_thread(os.stat, fspath)
                etag = self.etag(st)
                if etag not in [t.strip() for t in im.split(",")]:
                    raise falcon.HTTPPreconditionFailed()
            else:
                raise falcon.HTTPPreconditionFailed()

        cl = req.content_length
        if cl is not None and cl > MAX_UPLOAD_BYTES:
            raise falcon.HTTPPayloadTooLarge(
                description=f"Upload exceeds {MAX_UPLOAD_BYTES // (1024**3)} GiB limit"
            )
        existed = fspath.exists()

        # Stream the upload to a temp file in the hidden temp dir, then
        # os.replace into place — eliminates the buffer-the-whole-body
        # memory blow-up and the partial-write window where readers see a
        # half-finished file.
        tmp_path: pathlib.Path | None = None
        try:
            tmp = await asyncio.to_thread(
                tempfile.NamedTemporaryFile,
                dir=str(self.tmp_dir),
                delete=False,
                prefix="put-",
            )
            tmp_path = pathlib.Path(tmp.name)
            # NamedTemporaryFile uses 0o600 for safety, but the served tree
            # is by design world-readable (paths that aren't get hidden by
            # _world_visible). Mirror what pathlib.write_bytes produced
            # under the typical 022 umask so PUT-uploaded files keep
            # showing up in PROPFIND / GET.
            await asyncio.to_thread(os.chmod, tmp.name, 0o644)
            total = 0
            try:
                while True:
                    chunk = await req.bounded_stream.read(STREAM_CHUNK)
                    if not chunk:
                        break
                    total += len(chunk)
                    if total > MAX_UPLOAD_BYTES:
                        raise falcon.HTTPPayloadTooLarge(
                            description=(
                                f"Upload exceeds {MAX_UPLOAD_BYTES // (1024**3)} GiB"
                                " limit"
                            )
                        )
                    await asyncio.to_thread(tmp.write, chunk)
            finally:
                await asyncio.to_thread(tmp.close)

            # Best-effort If-Match TOCTOU close: re-stat right before
            # replace and re-validate. Sub-microsecond window remains, but
            # the seconds-wide pre-stream window is gone.
            if im and fspath.exists():
                st = await asyncio.to_thread(os.stat, fspath)
                etag_now = self.etag(st)
                if etag_now not in [t.strip() for t in im.split(",")]:
                    raise falcon.HTTPPreconditionFailed()

            try:
                await asyncio.to_thread(os.replace, str(tmp_path), str(fspath))
                tmp_path = None
            except OSError as e:
                if e.errno == errno.EXDEV:
                    raise falcon.HTTPInsufficientStorage(
                        description="Temp directory is on a different filesystem"
                    )
                raise
        finally:
            if tmp_path is not None:
                await asyncio.to_thread(_unlink_quiet, tmp_path)

        st = await asyncio.to_thread(os.stat, fspath)
        resp.set_header("ETag", self.etag(st))
        resp.status = falcon.HTTP_204 if existed else falcon.HTTP_201

    # -- DELETE ------------------------------------------------------------

    async def on_delete(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        fspath = self._resolve(kwargs.get("path", ""))
        if not fspath.exists():
            raise falcon.HTTPNotFound()
        self._check_lock(req, fspath)
        dav_path = self._dav_path(fspath)

        if fspath.is_dir():
            await asyncio.to_thread(shutil.rmtree, fspath)
        else:
            await asyncio.to_thread(fspath.unlink)

        self.lock_mgr.remove_for_path(dav_path)
        self.prop_mgr.delete(dav_path)
        resp.status = falcon.HTTP_204

    # -- MKCOL -------------------------------------------------------------

    async def on_mkcol(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        fspath = self._resolve(kwargs.get("path", ""))

        # MKCOL must not have a body (RFC 4918 §9.3.1)
        body = await _read_xml_body(req)
        if body:
            raise falcon.HTTPUnsupportedMediaType()

        if fspath.exists():
            raise falcon.HTTPMethodNotAllowed(ALL_METHODS.split(", "))
        if not fspath.parent.exists():
            raise falcon.HTTPConflict(description="Parent collection does not exist")
        await asyncio.to_thread(fspath.mkdir)
        resp.status = falcon.HTTP_201

    # -- COPY / MOVE shared preamble ----------------------------------------

    def _check_copy_move(
        self, req: falcon.asgi.Request, fspath: pathlib.Path
    ) -> tuple[pathlib.Path, bool]:
        """Validate and return (destination, existed_before)."""
        dst = self._parse_destination(req)
        fspath_r = fspath.resolve()
        dst_r = dst.resolve()
        if fspath_r == dst_r:
            raise falcon.HTTPForbidden(
                description="Source and destination are the same"
            )
        # Ancestor/descendant overlap: copying / moving a subtree into
        # itself (or its parent into itself) leads to recursion or wholesale
        # destruction. Block both directions.
        if dst_r.is_relative_to(fspath_r) or fspath_r.is_relative_to(dst_r):
            raise falcon.HTTPForbidden(
                description="Source and destination overlap"
            )
        if not dst.parent.exists():
            raise falcon.HTTPConflict(
                description="Destination parent does not exist"
            )
        overwrite = req.get_header("Overwrite") or "T"
        existed = dst.exists()
        if existed and overwrite.upper() == "F":
            raise falcon.HTTPPreconditionFailed()
        # File-vs-collection mismatches: refuse rather than silently
        # destroying the existing resource of the wrong type.
        if existed:
            if fspath.is_dir() and not dst.is_dir():
                raise falcon.HTTPConflict(
                    description="Cannot overwrite non-collection with collection"
                )
            if not fspath.is_dir() and dst.is_dir():
                raise falcon.HTTPConflict(
                    description="Cannot overwrite collection with non-collection"
                )
        return dst, existed

    def _make_tmp_path(self, prefix: str) -> pathlib.Path:
        return self.tmp_dir / f"{prefix}-{uuid.uuid4().hex}"

    # -- COPY --------------------------------------------------------------

    async def on_copy(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        fspath = self._resolve(kwargs.get("path", ""))
        if not fspath.exists():
            raise falcon.HTTPNotFound()
        depth = req.get_header("Depth") or "infinity"
        if depth not in {"0", "infinity"}:
            raise falcon.HTTPBadRequest(description="Invalid Depth value for COPY")
        dst, existed = self._check_copy_move(req, fspath)
        # Locks: source need not be unlocked (COPY does not modify it), but
        # the destination must be writable to the caller.
        self._check_lock(req, dst)

        src_dav = self._dav_path(fspath)
        dst_dav = self._dav_path(dst)

        try:
            if fspath.is_dir():
                if depth == "0":
                    if existed:
                        await asyncio.to_thread(shutil.rmtree, dst)
                    await asyncio.to_thread(dst.mkdir, parents=False)
                else:
                    tmp = self._make_tmp_path("copytree")
                    try:
                        await asyncio.to_thread(shutil.copytree, fspath, tmp)
                        if existed:
                            await asyncio.to_thread(shutil.rmtree, dst)
                        await asyncio.to_thread(os.rename, str(tmp), str(dst))
                        tmp = None
                    finally:
                        if tmp is not None:
                            await asyncio.to_thread(_rmtree_quiet, tmp)
            else:
                tmp = self._make_tmp_path("copy")
                try:
                    await asyncio.to_thread(shutil.copy2, fspath, tmp)
                    await asyncio.to_thread(os.replace, str(tmp), str(dst))
                    tmp = None
                finally:
                    if tmp is not None:
                        await asyncio.to_thread(_unlink_quiet, tmp)
        except OSError as e:
            if e.errno == errno.EXDEV:
                raise falcon.HTTPInsufficientStorage(
                    description="Destination is on a different filesystem"
                )
            raise

        # Dead properties are part of the resource state; carry them over.
        self.prop_mgr.copy(src_dav, dst_dav)
        resp.set_header("Content-Location", self._href(dst))
        resp.status = falcon.HTTP_204 if existed else falcon.HTTP_201

    # -- MOVE --------------------------------------------------------------

    async def on_move(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        fspath = self._resolve(kwargs.get("path", ""))
        if not fspath.exists():
            raise falcon.HTTPNotFound()
        self._check_lock(req, fspath)
        dst, existed = self._check_copy_move(req, fspath)
        self._check_lock(req, dst)

        if existed:
            if dst.is_dir():
                await asyncio.to_thread(shutil.rmtree, dst)
            else:
                await asyncio.to_thread(dst.unlink)

        try:
            await asyncio.to_thread(shutil.move, str(fspath), str(dst))
        except OSError as e:
            if e.errno == errno.EXDEV:
                raise falcon.HTTPInsufficientStorage(
                    description="Destination is on a different filesystem"
                )
            raise

        src_dav = self._dav_path(fspath)
        dst_dav = self._dav_path(dst)
        self.lock_mgr.move_locks(src_dav, dst_dav)
        self.prop_mgr.move(src_dav, dst_dav)
        resp.set_header("Content-Location", self._href(dst))
        resp.status = falcon.HTTP_204 if existed else falcon.HTTP_201

    # -- PROPFIND ----------------------------------------------------------

    async def on_propfind(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        fspath = self._resolve(kwargs.get("path", ""))
        if not fspath.exists():
            raise falcon.HTTPNotFound()

        depth = req.get_header("Depth") or "1"
        if depth not in {"0", "1", "infinity"}:
            raise falcon.HTTPBadRequest(description="Invalid Depth value")
        if depth == "infinity":
            raise falcon.HTTPForbidden(description="Depth: infinity not supported")

        body = await _read_xml_body(req)
        mode, requested = _parse_propfind(body)

        resources: list[pathlib.Path] = [fspath]
        if depth == "1" and fspath.is_dir():
            def _list_visible_children(d: pathlib.Path) -> list[pathlib.Path]:
                out = []
                for c in d.iterdir():
                    try:
                        if (not self.hide_dot_paths or not c.name.startswith(".")) and _world_visible(c.stat()):
                            out.append(c)
                    except OSError:
                        continue
                return sorted(out, key=lambda p: p.name)

            children = await asyncio.to_thread(_list_visible_children, fspath)
            resources.extend(children)

        viewer = getattr(req.context, "user", None)
        responses: list[etree._Element] = []
        for res in resources:
            try:
                st = await asyncio.to_thread(os.stat, res)
            except OSError:
                continue
            href = self._href(res)
            props_ok, props_404 = self._get_props(res, st, mode, requested, viewer)
            responses.append(_build_prop_response(href, props_ok, props_404))

        resp.status = "207 Multi-Status"
        resp.content_type = "application/xml; charset=utf-8"
        resp.data = _build_multistatus(responses)

    # -- PROPPATCH ---------------------------------------------------------

    async def on_proppatch(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        fspath = self._resolve(kwargs.get("path", ""))
        if not fspath.exists():
            raise falcon.HTTPNotFound()
        self._check_lock(req, fspath)

        body = await _read_xml_body(req)
        if not body:
            raise falcon.HTTPBadRequest(description="Empty PROPPATCH body")

        try:
            root = etree.fromstring(body, parser=_SECURE_PARSER)
        except etree.XMLSyntaxError:
            raise falcon.HTTPBadRequest(description="Invalid XML")

        dav_path = self._dav_path(fspath)
        props_ok: dict[str, etree._Element | str | None] = {}
        processed = 0

        for set_el in root.findall(f"{D}set"):
            prop_el = set_el.find(f"{D}prop")
            if prop_el is not None:
                for child in prop_el:
                    if processed >= MAX_PROP_CHILDREN:
                        raise falcon.HTTPBadRequest(
                            description="Too many properties in PROPPATCH"
                        )
                    processed += 1
                    key = child.tag
                    value = (
                        etree.tostring(child, encoding="unicode")
                        if len(child)
                        else (child.text or "")
                    )
                    self.prop_mgr.set_prop(dav_path, key, value)
                    props_ok[
                        child.tag.split("}")[-1] if "}" in child.tag else child.tag
                    ] = None

        for remove_el in root.findall(f"{D}remove"):
            prop_el = remove_el.find(f"{D}prop")
            if prop_el is not None:
                for child in prop_el:
                    if processed >= MAX_PROP_CHILDREN:
                        raise falcon.HTTPBadRequest(
                            description="Too many properties in PROPPATCH"
                        )
                    processed += 1
                    key = child.tag
                    self.prop_mgr.remove_prop(dav_path, key)
                    props_ok[
                        child.tag.split("}")[-1] if "}" in child.tag else child.tag
                    ] = None

        href = self._href(fspath)
        response = _build_prop_response(href, props_ok, [])
        resp.status = "207 Multi-Status"
        resp.content_type = "application/xml; charset=utf-8"
        resp.data = _build_multistatus([response])

    # -- LOCK --------------------------------------------------------------

    async def on_lock(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        fspath = self._resolve(kwargs.get("path", ""))
        dav_path = self._dav_path(fspath)
        depth = req.get_header("Depth") or "infinity"
        if depth not in {"0", "infinity"}:
            raise falcon.HTTPBadRequest(description="Invalid Depth value for LOCK")
        # Reject root + infinity — single request blocks every write to the
        # entire tree until expiry. Sub-tree infinity locks remain allowed.
        if dav_path == "/" and depth == "infinity":
            raise falcon.HTTPForbidden(
                description="Depth: infinity LOCK on / is not permitted"
            )
        timeout_secs = _parse_timeout(req.get_header("Timeout"))
        principal = getattr(req.context, "user", None)

        # Lock refresh
        if_header = req.get_header("If")
        if if_header:
            tokens = re.findall(r"<([^>]+)>", if_header)
            for token in tokens:
                lock = self.lock_mgr.refresh(token, principal, timeout_secs)
                if lock and lock.path == dav_path:
                    self._set_lock_response(resp, lock, falcon.HTTP_200)
                    return
            raise falcon.HTTPPreconditionFailed()

        body = await _read_xml_body(req)
        scope, owner_xml = _parse_lockinfo(body)
        remote_addr = req.access_route[-1] if req.access_route else req.remote_addr

        # Create the resource atomically if it does not exist (lock-null
        # resource). O_CREAT|O_EXCL closes the race that lets two concurrent
        # LOCKs both observe a missing file and both "create" it.
        created = False
        if not await asyncio.to_thread(fspath.exists):
            if not await asyncio.to_thread(fspath.parent.exists):
                raise falcon.HTTPConflict(
                    description="Parent collection does not exist"
                )
            try:
                await asyncio.to_thread(_atomic_touch, fspath)
                created = True
            except FileExistsError:
                created = False

        try:
            lock = self.lock_mgr.acquire(
                dav_path,
                owner_xml,
                depth,
                scope,
                timeout_secs,
                principal=principal,
                remote_addr=remote_addr,
            )
        except BaseException:
            if created:
                try:
                    await asyncio.to_thread(os.unlink, fspath)
                except OSError:
                    pass
            raise

        if not lock:
            if created:
                try:
                    await asyncio.to_thread(os.unlink, fspath)
                except OSError:
                    pass
            raise falcon.HTTPLocked()

        self._set_lock_response(
            resp, lock, falcon.HTTP_201 if created else falcon.HTTP_200
        )

    # -- UNLOCK ------------------------------------------------------------

    async def on_unlock(
        self, req: falcon.asgi.Request, resp: falcon.asgi.Response, **kwargs: str
    ) -> None:
        token_header = req.get_header("Lock-Token")
        if not token_header:
            raise falcon.HTTPBadRequest(description="Missing Lock-Token header")
        # RFC 4918 §10.5: Lock-Token uses the Coded-URL form: <token>.
        # Strip exactly one leading "<" and trailing ">"; reject if the body
        # still contains either character (defends against a greedy-strip
        # parser eating angle brackets that were part of the token).
        token = token_header.strip()
        if token.startswith("<") and token.endswith(">"):
            token = token[1:-1]
        if "<" in token or ">" in token:
            raise falcon.HTTPBadRequest(description="Malformed Lock-Token header")
        principal = getattr(req.context, "user", None)
        outcome = self.lock_mgr.release(token, principal)
        if outcome == "ok":
            resp.status = falcon.HTTP_204
        elif outcome == "forbidden":
            raise falcon.HTTPForbidden(
                description="Lock is held by a different principal"
            )
        else:
            raise falcon.HTTPConflict(description="Lock token not found")


# ---------------------------------------------------------------------------
# App factory
# ---------------------------------------------------------------------------


def _error_serializer(
    req: falcon.asgi.Request,
    resp: falcon.asgi.Response,
    exception: falcon.HTTPError,
) -> None:
    """Return plain-text error bodies instead of JSON."""
    resp.content_type = "text/plain; charset=utf-8"
    title = exception.status or "Error"
    desc = exception.description or ""
    body = f"{title}\n{desc}" if desc else str(title)
    resp.text = body


async def _generic_error_handler(
    req: falcon.asgi.Request,
    resp: falcon.asgi.Response,
    exception: BaseException,
    params: dict,
) -> None:
    """Catch-all for non-HTTPError exceptions. Returns plain 500 with no internals leaked."""
    if isinstance(exception, falcon.HTTPError):
        raise exception
    logging.exception("Unhandled exception serving %s %s", req.method, req.path)
    resp.status = falcon.HTTP_500
    resp.content_type = "text/plain; charset=utf-8"
    resp.text = "Internal Server Error"


def create_app(
    basedir: pathlib.Path,
    username: str | None = None,
    password: str | None = None,
    allow_anonymous_read: bool = False,
    hide_dot_paths: bool = True,
) -> falcon.asgi.App:
    if username and allow_anonymous_read:
        mode = "public reads, authenticated writes"
    elif username:
        mode = "all access requires auth"
    else:
        mode = "public read-only"

    # Hidden temp directory for atomic PUT / COPY / MOVE staging. Sweep on
    # every boot to clean up anything left behind by an interrupted server.
    tmp_dir = basedir / TMP_DIRNAME
    if tmp_dir.exists():
        shutil.rmtree(tmp_dir)
    tmp_dir.mkdir(mode=0o700)

    lock_mgr = LockManager()
    prop_mgr = PropertyManager()
    middleware: list = [AuthMiddleware(username, password, allow_anonymous_read)]
    app = falcon.asgi.App(middleware=middleware)
    app.set_error_serializer(_error_serializer)
    app.add_error_handler(Exception, _generic_error_handler)
    resource = WebDAVResource(basedir, lock_mgr, prop_mgr, mode, hide_dot_paths)
    app.add_route("/{path:path}", resource)
    return app


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

cli = typer.Typer(add_completion=False)


def _print_usage() -> None:
    lines = [
        "Usage: webdav.py --basedir DIR [OPTIONS]",
        "",
        "  RFC 4918 WebDAV server. Serves a directory over HTTP/WebDAV.",
        "  Secure by default: at least one of --allow-anonymous-read or",
        "  --username/--password must be provided.",
        "",
        "Options:",
        "  --basedir               TEXT  Directory to serve (required)",
        "  --port                  INT   Port to listen on [default: 8080]",
        "  --username              TEXT  Username for write access",
        "  --password              TEXT  Password for write access",
        "  --allow-anonymous-read        Allow unauthenticated reads",
        "  --help                        Show this message and exit",
        "",
        "Access matrix:",
        "  --username/--password only    → reads and writes require auth",
        "  --allow-anonymous-read only   → reads open, writes blocked (no creds)",
        "  both                          → reads open, writes require auth",
        "",
        "Examples:",
        "  # Public read-only share",
        "  uv run webdav.py --basedir . --allow-anonymous-read",
        "",
        "  # Private (all access requires auth)",
        "  uv run webdav.py --basedir /srv/files --username alice --password secret",
        "",
        "  # Public reads, authenticated writes",
        "  uv run webdav.py --basedir ./data --allow-anonymous-read \\",
        "                   --username alice --password secret",
        "",
        "macOS Finder:",
        "  Finder > Go > Connect to Server > http://<host>:<port>/",
    ]
    typer.echo("\n".join(lines))


@cli.command()
def main(
    basedir: str = typer.Option(
        None, help="Directory to serve. Must exist.", show_default=False
    ),
    port: int = typer.Option(8080, help="Port to listen on."),
    username: str | None = typer.Option(
        None, help="Username for write access. Requires --password."
    ),
    password: str | None = typer.Option(
        None, help="Password for write access. Requires --username."
    ),
    allow_anonymous_read: bool = typer.Option(
        False, "--allow-anonymous-read", help="Allow unauthenticated read access."
    ),
    show_hidden: bool = typer.Option(
        False, "--show-hidden", help="Expose dot-prefixed paths (default: hidden)."
    ),
    cert: str | None = typer.Option(
        None, "--cert", help="TLS certificate file (PEM). Pair with --key."
    ),
    key: str | None = typer.Option(
        None, "--key", help="TLS private-key file (PEM). Pair with --cert."
    ),
) -> None:
    """RFC 4918 WebDAV server. Serves a directory over HTTP/WebDAV.

    Secure by default: requires --username/--password and/or
    --allow-anonymous-read. Works with macOS Finder, Cyberduck, cadaver,
    and other standard WebDAV clients.
    """
    if not basedir:
        _print_usage()
        raise typer.Exit(0)
    # Credentials must be paired
    if bool(username) != bool(password):
        typer.echo(
            "Error: --username and --password must both be provided.\n"
            "\n"
            "  Private server:  uv run webdav.py --basedir DIR"
            " --username USER --password PASS\n"
            "  Public reads:    uv run webdav.py --basedir DIR"
            " --allow-anonymous-read",
            err=True,
        )
        raise typer.Exit(1)
    # Must enable at least one access mode
    if not username and not allow_anonymous_read:
        typer.echo(
            "Error: server would be inaccessible — provide credentials and/or"
            " --allow-anonymous-read.\n"
            "\n"
            "  Private server:  uv run webdav.py --basedir DIR"
            " --username USER --password PASS\n"
            "  Public reads:    uv run webdav.py --basedir DIR"
            " --allow-anonymous-read\n"
            "  Both:            uv run webdav.py --basedir DIR"
            " --allow-anonymous-read --username USER --password PASS",
            err=True,
        )
        raise typer.Exit(1)
    base = pathlib.Path(basedir).resolve()
    if not base.is_dir():
        typer.echo(
            f"Error: '{base}' is not a directory.\n"
            "\n"
            f"  Create it first:  mkdir -p {base}\n"
            f"  Or pick another:  uv run webdav.py --basedir /tmp/dav",
            err=True,
        )
        raise typer.Exit(1)
    app = create_app(base, username, password, allow_anonymous_read, hide_dot_paths=not show_hidden)
    if username and allow_anonymous_read:
        mode = "public reads, authenticated writes"
    elif username:
        mode = "all access requires auth"
    else:
        mode = "public read-only"
    if bool(cert) != bool(key):
        typer.echo("Error: --cert and --key must both be provided.", err=True)
        raise typer.Exit(1)
    if not cert and username:
        typer.echo(
            "WARNING: server is configured for Basic Auth over plain HTTP. "
            "Use --cert/--key or terminate TLS upstream to protect credentials.",
            err=True,
        )
    scheme = "https" if cert else "http"
    typer.echo(
        f"WebDAV server [{mode}] on {scheme}://0.0.0.0:{port}/ serving {base}"
    )
    uvicorn_kwargs: dict = {"host": "0.0.0.0", "port": port}
    if cert:
        uvicorn_kwargs["ssl_certfile"] = cert
        uvicorn_kwargs["ssl_keyfile"] = key
    uvicorn.run(app, **uvicorn_kwargs)


if __name__ == "__main__":
    cli()