FILE: reportkit/__init__.py
```python
"""reportkit — multi-tenant report service."""
```

FILE: reportkit/schema.sql
```sql
-- reportkit schema
-- All DDL is idempotent (IF NOT EXISTS).

PRAGMA journal_mode = WAL;
PRAGMA foreign_keys = ON;

CREATE TABLE IF NOT EXISTS tenants (
    id          TEXT PRIMARY KEY,          -- UUID
    suspended   INTEGER NOT NULL DEFAULT 0 CHECK (suspended IN (0, 1)),
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE TABLE IF NOT EXISTS users (
    id          TEXT PRIMARY KEY,          -- UUID
    tenant_id   TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    role        TEXT NOT NULL CHECK (role IN ('viewer', 'editor', 'admin')),
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE INDEX IF NOT EXISTS idx_users_tenant ON users(tenant_id);

CREATE TABLE IF NOT EXISTS sessions (
    token_hash  TEXT PRIMARY KEY,          -- SHA-256 hex of the bearer token
    user_id     TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    expires_at  TEXT NOT NULL,             -- ISO-8601 UTC
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);

CREATE TABLE IF NOT EXISTS reports (
    id          TEXT PRIMARY KEY,          -- UUID
    tenant_id   TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    owner_id    TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    title       TEXT NOT NULL CHECK (length(title) BETWEEN 1 AND 500),
    body        TEXT NOT NULL DEFAULT '',
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    updated_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE INDEX IF NOT EXISTS idx_reports_tenant_owner ON reports(tenant_id, owner_id);
CREATE INDEX IF NOT EXISTS idx_reports_created      ON reports(tenant_id, created_at DESC);

CREATE TABLE IF NOT EXISTS attachments (
    id          TEXT PRIMARY KEY,          -- UUID
    tenant_id   TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    owner_id    TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    report_id   TEXT REFERENCES reports(id) ON DELETE SET NULL,
    filename    TEXT NOT NULL CHECK (length(filename) BETWEEN 1 AND 255),
    size_bytes  INTEGER NOT NULL CHECK (size_bytes >= 0),
    scan_status TEXT NOT NULL DEFAULT 'pending'
                     CHECK (scan_status IN ('pending', 'clean', 'rejected')),
    storage_key TEXT NOT NULL UNIQUE,      -- server-generated UUID path
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE INDEX IF NOT EXISTS idx_attachments_tenant ON attachments(tenant_id);

CREATE TABLE IF NOT EXISTS invoice_events (
    event_id    TEXT PRIMARY KEY,          -- idempotency key from provider
    tenant_id   TEXT NOT NULL,
    amount      INTEGER NOT NULL,          -- cents
    currency    TEXT NOT NULL CHECK (length(currency) = 3),
    processed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE TABLE IF NOT EXISTS audit_log (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    ts          TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    actor_id    TEXT,                      -- NULL for system/webhook events
    tenant_id   TEXT,
    action      TEXT NOT NULL,
    target_type TEXT,
    target_id   TEXT,
    detail      TEXT                       -- JSON, no PII/secrets
);

CREATE INDEX IF NOT EXISTS idx_audit_actor  ON audit_log(actor_id);
CREATE INDEX IF NOT EXISTS idx_audit_tenant ON audit_log(tenant_id);
CREATE INDEX IF NOT EXISTS idx_audit_ts     ON audit_log(ts);
```

FILE: reportkit/errors.py
```python
"""Typed error hierarchy for reportkit."""


class ReportkitError(Exception):
    """Base class."""
    http_status: int = 500
    code: str = "internal_error"


class BadRequest(ReportkitError):
    http_status = 400
    code = "bad_request"


class Unauthorized(ReportkitError):
    http_status = 401
    code = "unauthorized"


class Forbidden(ReportkitError):
    http_status = 403
    code = "forbidden"


class NotFound(ReportkitError):
    http_status = 404
    code = "not_found"


class Conflict(ReportkitError):
    http_status = 409
    code = "conflict"


class TenantSuspended(ReportkitError):
    http_status = 403
    code = "tenant_suspended"


class RateLimited(ReportkitError):
    http_status = 429
    code = "rate_limited"


class PayloadTooLarge(ReportkitError):
    http_status = 413
    code = "payload_too_large"


class UnsupportedMediaType(ReportkitError):
    http_status = 415
    code = "unsupported_media_type"


class WebhookAuthFailed(ReportkitError):
    http_status = 401
    code = "webhook_auth_failed"
```

FILE: reportkit/config.py
```python
"""
Runtime configuration.

All secrets come from the environment; defaults are safe-fail values
(empty strings that will cause auth to fail rather than silently pass).
"""

import os


class Config:
    # Billing webhook HMAC secret — must be set in production.
    BILLING_WEBHOOK_SECRET: bytes = os.environ.get(
        "BILLING_WEBHOOK_SECRET", ""
    ).encode()

    # Webhook timestamp tolerance in seconds (replay-window).
    WEBHOOK_TIMESTAMP_TOLERANCE: int = int(
        os.environ.get("WEBHOOK_TIMESTAMP_TOLERANCE", "300")
    )

    # Maximum attachment size in bytes (25 MiB).
    MAX_ATTACHMENT_BYTES: int = int(
        os.environ.get("MAX_ATTACHMENT_BYTES", str(25 * 1024 * 1024))
    )

    # Session token TTL in seconds (8 hours).
    SESSION_TTL_SECONDS: int = int(
        os.environ.get("SESSION_TTL_SECONDS", str(8 * 3600))
    )

    # Rate-limit: max requests per window per (tenant, endpoint) key.
    RATE_LIMIT_WINDOW_SECONDS: int = int(
        os.environ.get("RATE_LIMIT_WINDOW_SECONDS", "60")
    )
    RATE_LIMIT_MAX_REQUESTS: int = int(
        os.environ.get("RATE_LIMIT_MAX_REQUESTS", "120")
    )

    # Allowed MIME types for attachments.
    ALLOWED_ATTACHMENT_MIME_TYPES: frozenset = frozenset(
        os.environ.get(
            "ALLOWED_ATTACHMENT_MIME_TYPES",
            "application/pdf,image/png,image/jpeg,image/gif,"
            "text/plain,application/zip",
        ).split(",")
    )

    # Disallowed content markers (byte strings) for the attachment scanner.
    # In production this would delegate to a real AV library.
    DISALLOWED_CONTENT_MARKERS: tuple = (
        b"EICAR-STANDARD-ANTIVIRUS-TEST-FILE",
        b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR",
    )
```

FILE: reportkit/db.py
```python
"""
Database helpers.

All queries use parameterised statements.  Identifiers that must be
interpolated (ORDER BY column) are mapped through a hard-coded allowlist.
"""

import sqlite3
import os
from pathlib import Path


def open_db(path: str = ":memory:") -> sqlite3.Connection:
    """
    Open (or create) the SQLite database and apply the schema.

    WAL mode and foreign-key enforcement are set on every connection.
    Row factory is set so callers get dict-like rows.
    """
    conn = sqlite3.connect(path, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode = WAL")
    conn.execute("PRAGMA foreign_keys = ON")

    schema_path = Path(__file__).with_name("schema.sql")
    conn.executescript(schema_path.read_text())
    conn.commit()
    return conn


# ---------------------------------------------------------------------------
# Allowlisted sort columns for ORDER BY injection prevention (rules/01 §2).
# Maps the user-visible name to the exact SQL column expression.
# ---------------------------------------------------------------------------
REPORT_SORT_COLUMNS: dict[str, str] = {
    "created_at": "created_at",
    "title": "title",
}

DEFAULT_SORT_COLUMN = "created_at"
```

FILE: reportkit/auth.py
```python
"""
Authentication and permission resolution.

Token → session → user → role → permission set.

The permission cache is an in-process dict keyed by (user_id, role).
Roles are changed from the ops console; the cache entry is invalidated
when set_role() is called.  TTL is a safety net.
"""

import hashlib
import time
import sqlite3
from typing import Optional

from .errors import Unauthorized, Forbidden, TenantSuspended
from .logging import get_logger

logger = get_logger(__name__)

# ---------------------------------------------------------------------------
# Permission model
# ---------------------------------------------------------------------------

_ROLE_PERMISSIONS: dict[str, frozenset] = {
    "viewer": frozenset({"report:read"}),
    "editor": frozenset({"report:read", "report:write"}),
    "admin":  frozenset({"report:read", "report:write",
                         "report:delete", "user:manage"}),
}


def permissions_for_role(role: str) -> frozenset:
    return _ROLE_PERMISSIONS.get(role, frozenset())


# ---------------------------------------------------------------------------
# Token hashing — SHA-256 is fine for high-entropy bearer tokens (rules/02 §8).
# ---------------------------------------------------------------------------

def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


# ---------------------------------------------------------------------------
# Session resolution
# ---------------------------------------------------------------------------

class Principal:
    """Resolved, authenticated caller."""

    __slots__ = ("user_id", "tenant_id", "role", "permissions")

    def __init__(self, user_id: str, tenant_id: str, role: str):
        self.user_id = user_id
        self.tenant_id = tenant_id
        self.role = role
        self.permissions: frozenset = permissions_for_role(role)

    def require(self, permission: str) -> None:
        """Raise Forbidden if the caller lacks *permission*."""
        if permission not in self.permissions:
            raise Forbidden(
                f"Permission '{permission}' required; caller has role '{self.role}'"
            )

    def __repr__(self) -> str:  # pragma: no cover
        return (
            f"Principal(user_id={self.user_id!r}, "
            f"tenant_id={self.tenant_id!r}, role={self.role!r})"
        )


def resolve_principal(db: sqlite3.Connection, token: str) -> Principal:
    """
    Resolve a bearer token to a Principal.

    Raises Unauthorized on any failure so callers cannot distinguish
    "no such token" from "expired" (account enumeration, rules/02 §1).
    """
    if not token or not isinstance(token, str):
        raise Unauthorized("Missing bearer token")

    # Length-cap: tokens are UUIDs or similar; reject absurdly long strings.
    if len(token) > 512:
        raise Unauthorized("Token too long")

    token_hash = hash_token(token)
    now_iso = _utcnow_iso()

    row = db.execute(
        """
        SELECT s.user_id, s.expires_at,
               u.tenant_id, u.role,
               t.suspended
        FROM   sessions s
        JOIN   users   u ON u.id = s.user_id
        JOIN   tenants t ON t.id = u.tenant_id
        WHERE  s.token_hash = ?
        """,
        (token_hash,),
    ).fetchone()

    if row is None:
        logger.warning("auth_failure", extra={"reason": "token_not_found"})
        raise Unauthorized("Invalid or expired token")

    if row["expires_at"] < now_iso:
        logger.warning("auth_failure", extra={"reason": "token_expired"})
        raise Unauthorized("Invalid or expired token")

    if row["suspended"]:
        raise TenantSuspended("Tenant is suspended")

    return Principal(
        user_id=row["user_id"],
        tenant_id=row["tenant_id"],
        role=row["role"],
    )


def _utcnow_iso() -> str:
    import datetime
    return datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z"
```

FILE: reportkit/rate_limit.py
```python
"""
In-process token-bucket rate limiter keyed on (tenant_id, endpoint).

For a production deployment this would be backed by Redis so limits are
shared across processes.  The interface is identical; swap the store.

Rules/05 non-negotiable (a): abuse control — rate limiting keyed to the caller.
"""

import time
import threading
from collections import defaultdict

from .errors import RateLimited
from .logging import get_logger

logger = get_logger(__name__)


class _Bucket:
    __slots__ = ("count", "window_start")

    def __init__(self, window_start: float):
        self.count = 0
        self.window_start = window_start


class RateLimiter:
    """
    Fixed-window counter per (tenant_id, endpoint).

    Thread-safe via a per-instance lock.
    """

    def __init__(self, window_seconds: int = 60, max_requests: int = 120):
        self._window = window_seconds
        self._max = max_requests
        self._buckets: dict[tuple, _Bucket] = defaultdict(
            lambda: _Bucket(time.monotonic())
        )
        self._lock = threading.Lock()

    def check(self, tenant_id: str, endpoint: str) -> None:
        """
        Increment the counter for (tenant_id, endpoint).

        Raises RateLimited if the window quota is exceeded.
        """
        key = (tenant_id, endpoint)
        now = time.monotonic()

        with self._lock:
            bucket = self._buckets[key]
            if now - bucket.window_start >= self._window:
                bucket.count = 0
                bucket.window_start = now
            bucket.count += 1
            if bucket.count > self._max:
                logger.warning(
                    "rate_limit_exceeded",
                    extra={"tenant_id": tenant_id, "endpoint": endpoint},
                )
                raise RateLimited(
                    f"Rate limit exceeded for endpoint '{endpoint}'"
                )
```

FILE: reportkit/validation.py
```python
"""
Input validation helpers.

All validation is allowlist-based, performed after any decoding,
server-side (rules/01 §1).
"""

import re
import unicodedata
from typing import Any

from .errors import BadRequest

# ---------------------------------------------------------------------------
# Field-level validators
# ---------------------------------------------------------------------------

# Printable Unicode text: letters, numbers, common punctuation, spaces.
# Rejects NUL, control characters, and surrogate code points.
_SAFE_TEXT_RE = re.compile(
    r"^[\u0020-\uD7FF\uE000-\uFFFD\U00010000-\U0010FFFF]{1,500}$"
)

# Title: 1–500 printable characters, no NUL.
_TITLE_RE = re.compile(
    r"^[\u0020-\uD7FF\uE000-\uFFFD\U00010000-\U0010FFFF]{1,500}$"
)

# Filename: printable ASCII only, no path separators, no NUL, no control chars.
# Allowlist: letters, digits, hyphen, underscore, dot, space.
_FILENAME_RE = re.compile(r"^[A-Za-z0-9 ._\-]{1,255}$")

# UUID v4 pattern.
_UUID_RE = re.compile(
    r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$",
    re.IGNORECASE,
)

# Allowed sort keys — validated against the DB allowlist in db.py too.
_ALLOWED_SORT_KEYS = frozenset({"created_at", "title"})

# Allowed MIME type pattern (type/subtype, no parameters).
_MIME_RE = re.compile(r"^[a-z0-9][a-z0-9!#$&\-^_]{0,126}/[a-z0-9][a-z0-9!#$&\-^_.+]{0,126}$")


def require_string(value: Any, field: str, min_len: int = 1, max_len: int = 500) -> str:
    """Assert *value* is a non-empty string within length bounds."""
    if not isinstance(value, str):
        raise BadRequest(f"'{field}' must be a string")
    # Canonicalize: NFC normalization before validation (rules/01 §1).
    value = unicodedata.normalize("NFC", value)
    if "\x00" in value:
        raise BadRequest(f"'{field}' must not contain NUL bytes")
    if not (min_len <= len(value) <= max_len):
        raise BadRequest(
            f"'{field}' must be between {min_len} and {max_len} characters"
        )
    return value


def require_title(value: Any) -> str:
    s = require_string(value, "title", 1, 500)
    if not _TITLE_RE.fullmatch(s):
        raise BadRequest("'title' contains disallowed characters")
    return s


def require_body(value: Any) -> str:
    """Body may be empty but must be a string and ≤ 1 MiB."""
    if not isinstance(value, str):
        raise BadRequest("'body' must be a string")
    value = unicodedata.normalize("NFC", value)
    if "\x00" in value:
        raise BadRequest("'body' must not contain NUL bytes")
    if len(value) > 1_048_576:
        raise BadRequest("'body' exceeds 1 MiB limit")
    return value


def require_uuid(value: Any, field: str = "id") -> str:
    if not isinstance(value, str):
        raise BadRequest(f"'{field}' must be a string")
    value = value.strip().lower()
    if not _UUID_RE.fullmatch(value):
        raise BadRequest(f"'{field}' is not a valid UUID")
    return value


def require_sort_key(value: Any) -> str:
    """Validate a user-supplied sort key against the allowlist."""
    if not isinstance(value, str):
        raise BadRequest("'sort' must be a string")
    value = value.strip().lower()
    if value not in _ALLOWED_SORT_KEYS:
        raise BadRequest(
            f"'sort' must be one of: {', '.join(sorted(_ALLOWED_SORT_KEYS))}"
        )
    return value


def require_filename(value: Any) -> str:
    if not isinstance(value, str):
        raise BadRequest("'filename' must be a string")
    value = unicodedata.normalize("NFC", value)
    if "\x00" in value:
        raise BadRequest("'filename' must not contain NUL bytes")
    # Reject path traversal components.
    if ".." in value or "/" in value or "\\" in value or ":" in value:
        raise BadRequest("'filename' must not contain path separators or colons")
    if not _FILENAME_RE.fullmatch(value):
        raise BadRequest(
            "'filename' must contain only letters, digits, spaces, "
            "hyphens, underscores, and dots"
        )
    return value


def require_mime_type(value: Any) -> str:
    if not isinstance(value, str):
        raise BadRequest("'content_type' must be a string")
    value = value.strip().lower()
    if not _MIME_RE.fullmatch(value):
        raise BadRequest("'content_type' is not a valid MIME type")
    return value


def require_bytes(value: Any, field: str, max_bytes: int) -> bytes:
    if not isinstance(value, (bytes, bytearray)):
        raise BadRequest(f"'{field}' must be bytes")
    if len(value) > max_bytes:
        raise BadRequest(
            f"'{field}' exceeds maximum size of {max_bytes} bytes"
        )
    return bytes(value)


def extract_bearer_token(request: dict) -> str:
    """
    Extract the bearer token from request["headers"]["Authorization"].

    Raises Unauthorized (via the caller) if absent or malformed.
    """
    from .errors import Unauthorized

    headers = request.get("headers") or {}
    auth = headers.get("Authorization") or headers.get("authorization") or ""
    if not isinstance(auth, str):
        raise Unauthorized("Authorization header must be a string")
    auth = auth.strip()
    if not auth.lower().startswith("bearer "):
        raise Unauthorized("Authorization header must use Bearer scheme")
    token = auth[7:].strip()
    if not token:
        raise Unauthorized("Bearer token is empty")
    return token
```

FILE: reportkit/logging.py
```python
"""
Structured logging without secrets or PII (rules/05 non-negotiable (d)).

All log records are JSON lines on stderr.  Fields that must never appear:
  - bearer tokens, session hashes, passwords, HMAC secrets
  - full email addresses, phone numbers, or other PII

The logger wraps stdlib logging with a JSON formatter.
"""

import json
import logging
import sys
import time
from typing import Any


class _JsonFormatter(logging.Formatter):
    """Emit each record as a single JSON line."""

    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        # Merge any extra fields the caller passed.
        for key, val in record.__dict__.items():
            if key not in (
                "args", "created", "exc_info", "exc_text", "filename",
                "funcName", "levelname", "levelno", "lineno", "message",
                "module", "msecs", "msg", "name", "pathname", "process",
                "processName", "relativeCreated", "stack_info", "taskName",
                "thread", "threadName",
            ):
                payload[key] = val
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload, default=str)


def _configure_root() -> None:
    handler = logging.StreamHandler(sys.stderr)
    handler.setFormatter(_JsonFormatter())
    root = logging.getLogger()
    if not root.handlers:
        root.addHandler(handler)
    root.setLevel(logging.INFO)


_configure_root()


def get_logger(name: str) -> logging.Logger:
    return logging.getLogger(name)
```

FILE: reportkit/events.py
```python
"""
Domain event dispatcher.

Handlers register against an event name.  The dispatcher calls every
registered handler in registration order.  Handler exceptions are caught
and logged so one bad handler cannot abort the others.

The `tenant.quota_exceeded` handler is registered here as a stub.
The batch-import feed that emits it was retired; the handler is wired
but will not fire until that feed is restored.  It is NOT called from
any live code path.  The platform team has been notified (see NOTES.md).
"""

import threading
from typing import Callable, Any

from .logging import get_logger

logger = get_logger(__name__)

_registry: dict[str, list[Callable]] = {}
_lock = threading.Lock()


def register(event_name: str, handler: Callable) -> None:
    """Register *handler* for *event_name*."""
    with _lock:
        _registry.setdefault(event_name, []).append(handler)


def dispatch(event_name: str, payload: dict[str, Any]) -> None:
    """
    Call every handler registered for *event_name*.

    Exceptions in individual handlers are caught and logged; they do not
    propagate to the caller.
    """
    with _lock:
        handlers = list(_registry.get(event_name, []))

    for handler in handlers:
        try:
            handler(payload)
        except Exception:
            logger.exception(
                "event_handler_error",
                extra={"event": event_name, "handler": handler.__qualname__},
            )


# ---------------------------------------------------------------------------
# Built-in handlers
# ---------------------------------------------------------------------------

def _on_report_created(payload: dict) -> None:
    logger.info(
        "event.report.created",
        extra={
            "report_id": payload.get("report_id"),
            "tenant_id": payload.get("tenant_id"),
        },
    )


def _on_report_deleted(payload: dict) -> None:
    logger.info(
        "event.report.deleted",
        extra={
            "report_id": payload.get("report_id"),
            "tenant_id": payload.get("tenant_id"),
        },
    )


def _on_tenant_quota_exceeded(payload: dict) -> None:
    """
    Suspend a tenant when their quota is exceeded.

    This handler is registered but the emitting feed (batch-import) is
    currently retired.  It will activate automatically when the feed is
    restored and begins emitting `tenant.quota_exceeded` events.
    """
    logger.warning(
        "event.tenant.quota_exceeded",
        extra={"tenant_id": payload.get("tenant_id")},
    )
    # The db reference is injected via payload by the caller when the feed
    # is restored.  Guard defensively.
    db = payload.get("db")
    tenant_id = payload.get("tenant_id")
    if db is not None and tenant_id:
        try:
            db.execute(
                "UPDATE tenants SET suspended = 1 WHERE id = ?",
                (tenant_id,),
            )
            db.commit()
            logger.warning(
                "tenant_suspended_quota",
                extra={"tenant_id": tenant_id},
            )
        except Exception:
            logger.exception(
                "tenant_suspend_failed",
                extra={"tenant_id": tenant_id},
            )


register("report.created", _on_report_created)
register("report.deleted", _on_report_deleted)
register("tenant.quota_exceeded", _on_tenant_quota_exceeded)
```

FILE: reportkit/scanner.py
```python
"""
Attachment content scanner.

Checks uploaded bytes for disallowed content markers before storage.
In production this would delegate to a real AV/DLP library; the
interface is stable so the implementation can be swapped.
"""

from .config import Config
from .logging import get_logger

logger = get_logger(__name__)


def scan(data: bytes, filename: str) -> tuple[bool, str]:
    """
    Scan *data* for disallowed content.

    Returns (is_clean: bool, reason: str).
    reason is empty when is_clean is True.
    """
    for marker in Config.DISALLOWED_CONTENT_MARKERS:
        if marker in data:
            logger.warning(
                "attachment_scan_rejected",
                extra={"filename": filename, "marker_prefix": marker[:8].hex()},
            )
            return False, "disallowed content detected"

    logger.info(
        "attachment_scan_clean",
        extra={"filename": filename, "size": len(data)},
    )
    return True, ""
```

FILE: reportkit/audit.py
```python
"""
Immutable audit log writer (rules/03 §7).

Every admin mutation, impersonation, and security event is recorded here.
The detail field is JSON; it must never contain secrets or PII.
"""

import json
import sqlite3
from typing import Any, Optional

from .logging import get_logger

logger = get_logger(__name__)


def record(
    db: sqlite3.Connection,
    action: str,
    *,
    actor_id: Optional[str] = None,
    tenant_id: Optional[str] = None,
    target_type: Optional[str] = None,
    target_id: Optional[str] = None,
    detail: Optional[dict[str, Any]] = None,
) -> None:
    """
    Write one audit record.

    Failures are logged but do not raise — a failed audit write must not
    abort the business operation, but it must be visible.
    """
    detail_json = json.dumps(detail, default=str) if detail else None
    try:
        db.execute(
            """
            INSERT INTO audit_log
                (actor_id, tenant_id, action, target_type, target_id, detail)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (actor_id, tenant_id, action, target_type, target_id, detail_json),
        )
        # Callers commit their own transactions; audit rows ride the same tx.
    except Exception:
        logger.exception(
            "audit_write_failed",
            extra={
                "action": action,
                "actor_id": actor_id,
                "tenant_id": tenant_id,
            },
        )
```

FILE: reportkit/ops.py
```python
"""
Internal ops-console operations.

These functions are called by the internal ops console, which runs inside
the private network.  They are NOT exposed over HTTP.

Both functions write audit records before committing so the trail is
atomic with the mutation.
"""

import sqlite3

from .audit import record as audit_record
from .errors import NotFound, BadRequest
from .logging import get_logger
from .validation import require_uuid

logger = get_logger(__name__)

_VALID_ROLES = frozenset({"viewer", "editor", "admin"})


def purge_tenant(db: sqlite3.Connection, actor_id: str, tenant_id: str) -> int:
    """
    Delete every report belonging to *tenant_id*.

    Returns the number of reports deleted.
    Raises NotFound if the tenant does not exist.
    """
    require_uuid(actor_id, "actor_id")
    require_uuid(tenant_id, "tenant_id")

    row = db.execute(
        "SELECT id FROM tenants WHERE id = ?", (tenant_id,)
    ).fetchone()
    if row is None:
        raise NotFound(f"Tenant {tenant_id!r} not found")

    result = db.execute(
        "DELETE FROM reports WHERE tenant_id = ?", (tenant_id,)
    )
    deleted = result.rowcount

    audit_record(
        db,
        action="tenant.purge_reports",
        actor_id=actor_id,
        tenant_id=tenant_id,
        detail={"deleted_count": deleted},
    )
    db.commit()

    logger.info(
        "tenant_purged",
        extra={
            "actor_id": actor_id,
            "tenant_id": tenant_id,
            "deleted_count": deleted,
        },
    )
    return deleted


def set_role(
    db: sqlite3.Connection,
    actor_id: str,
    tenant_id: str,
    user_id: str,
    new_role: str,
) -> None:
    """
    Change *user_id*'s role to *new_role* within *tenant_id*.

    Raises NotFound if the user does not exist in that tenant.
    Raises BadRequest if *new_role* is not a valid role.
    """
    require_uuid(actor_id, "actor_id")
    require_uuid(tenant_id, "tenant_id")
    require_uuid(user_id, "user_id")

    if new_role not in _VALID_ROLES:
        raise BadRequest(
            f"Invalid role {new_role!r}; must be one of "
            f"{', '.join(sorted(_VALID_ROLES))}"
        )

    row = db.execute(
        "SELECT role FROM users WHERE id = ? AND tenant_id = ?",
        (user_id, tenant_id),
    ).fetchone()
    if row is None:
        raise NotFound(f"User {user_id!r} not found in tenant {tenant_id!r}")

    old_role = row["role"]
    db.execute(
        "UPDATE users SET role = ? WHERE id = ? AND tenant_id = ?",
        (new_role, user_id, tenant_id),
    )

    audit_record(
        db,
        action="user.set_role",
        actor_id=actor_id,
        tenant_id=tenant_id,
        target_type="user",
        target_id=user_id,
        detail={"old_role": old_role, "new_role": new_role},
    )
    db.commit()

    logger.info(
        "role_changed",
        extra={
            "actor_id": actor_id,
            "tenant_id": tenant_id,
            "user_id": user_id,
            "old_role": old_role,
            "new_role": new_role,
        },
    )
```

FILE: reportkit/webhook.py
```python
"""
Billing webhook handler.

Verifies the HMAC-SHA256 signature over the raw body before any parsing.
Implements idempotency via the invoice_events table.

Security controls (rules/02 §8, rules/05 non-negotiables):
  - Signature verified constant-time over the raw body before parsing.
  - Timestamp window enforced to prevent replay.
  - Idempotency key stored to prevent double-processing on provider retries.
  - No secret or PII in logs.
"""

import hashlib
import hmac
import json
import sqlite3
import time
from typing import Any

from .audit import record as audit_record
from .config import Config
from .errors import BadRequest, Conflict, WebhookAuthFailed
from .logging import get_logger
from .validation import require_uuid, require_string

logger = get_logger(__name__)


def handle_billing_webhook(
    db: sqlite3.Connection,
    raw_body: bytes,
    signature_header: str,
    timestamp_header: str,
) -> dict[str, Any]:
    """
    Process an `invoice.paid` webhook from the billing provider.

    Parameters
    ----------
    db:
        Open database connection.
    raw_body:
        The exact bytes received from the provider (before any decoding).
    signature_header:
        Value of the provider's signature header (hex-encoded HMAC-SHA256).
    timestamp_header:
        Value of the provider's timestamp header (Unix epoch as a string).

    Returns a dict with the processing result.
    Raises WebhookAuthFailed, BadRequest, or Conflict on error.
    """
    # --- 1. Authenticate BEFORE parsing (rules/03 §1, rules/02 §8) ----------
    _verify_signature(raw_body, signature_header, timestamp_header)

    # --- 2. Parse only after auth -------------------------------------------
    try:
        payload = json.loads(raw_body)
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        raise BadRequest(f"Invalid JSON body: {exc}") from exc

    if not isinstance(payload, dict):
        raise BadRequest("Webhook body must be a JSON object")

    event_type = payload.get("type")
    if event_type != "invoice.paid":
        # Unknown event types are acknowledged but not processed.
        logger.info("webhook_ignored", extra={"event_type": event_type})
        return {"status": "ignored", "event_type": event_type}

    # --- 3. Validate fields --------------------------------------------------
    event_id = payload.get("id")
    if not isinstance(event_id, str) or not event_id:
        raise BadRequest("'id' field is required and must be a non-empty string")
    # Length-cap the event_id (it's a provider-generated opaque string).
    if len(event_id) > 128:
        raise BadRequest("'id' field too long")

    data = payload.get("data") or {}
    if not isinstance(data, dict):
        raise BadRequest("'data' must be an object")

    tenant_id = data.get("tenant_id")
    require_uuid(tenant_id, "tenant_id")

    amount = data.get("amount")
    if not isinstance(amount, int) or amount <= 0:
        raise BadRequest("'amount' must be a positive integer (cents)")

    currency = data.get("currency")
    if not isinstance(currency, str) or len(currency) != 3 or not currency.isalpha():
        raise BadRequest("'currency' must be a 3-letter ISO 4217 code")
    currency = currency.upper()

    # --- 4. Idempotency: reject duplicate event_ids -------------------------
    existing = db.execute(
        "SELECT event_id FROM invoice_events WHERE event_id = ?",
        (event_id,),
    ).fetchone()
    if existing is not None:
        logger.info("webhook_duplicate", extra={"event_id": event_id})
        return {"status": "duplicate", "event_id": event_id}

    # --- 5. Persist ----------------------------------------------------------
    db.execute(
        """
        INSERT INTO invoice_events (event_id, tenant_id, amount, currency)
        VALUES (?, ?, ?, ?)
        """,
        (event_id, tenant_id, amount, currency),
    )
    audit_record(
        db,
        action="billing.invoice_paid",
        tenant_id=tenant_id,
        detail={"event_id": event_id, "currency": currency},
        # amount deliberately omitted from audit detail to avoid financial PII
        # in the general audit log; it lives in invoice_events.
    )
    db.commit()

    logger.info(
        "webhook_processed",
        extra={"event_id": event_id, "tenant_id": tenant_id},
    )
    return {"status": "ok", "event_id": event_id}


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------

def _verify_signature(
    raw_body: bytes,
    signature_header: str,
    timestamp_header: str,
) -> None:
    """
    Verify HMAC-SHA256 signature and timestamp window.

    Raises WebhookAuthFailed on any failure.
    The error message is deliberately generic (no oracle).
    """
    secret = Config.BILLING_WEBHOOK_SECRET
    if not secret:
        logger.error("webhook_secret_not_configured")
        raise WebhookAuthFailed("Webhook authentication not configured")

    # Validate timestamp to prevent replay (rules/02 §8).
    try:
        ts = int(timestamp_header)
    except (ValueError, TypeError):
        raise WebhookAuthFailed("Invalid webhook request")

    now = int(time.time())
    if abs(now - ts) > Config.WEBHOOK_TIMESTAMP_TOLERANCE:
        logger.warning("webhook_replay_rejected", extra={"delta": abs(now - ts)})
        raise WebhookAuthFailed("Invalid webhook request")

    # Compute expected signature: HMAC-SHA256(secret, timestamp + "." + body).
    # The signed payload includes the timestamp so it cannot be replayed with
    # a different timestamp.
    signed_payload = f"{ts}.".encode() + raw_body
    expected = hmac.new(secret, signed_payload, hashlib.sha256).hexdigest()

    # Constant-time comparison (rules/02 §8).
    if not hmac.compare_digest(expected, signature_header.strip()):
        logger.warning("webhook_signature_mismatch")
        raise WebhookAuthFailed("Invalid webhook request")
```

FILE: reportkit/service.py
```python
"""
ReportService — the HTTP surface.

Each public method takes a `request` dict and returns a response dict.
Authentication, authorization, rate-limiting, and input validation are
applied at the top of every handler before any business logic.

Request dict shape (callers supply what they need):
    {
        "headers": {"Authorization": "Bearer <token>", ...},
        "body":    {...},          # parsed JSON
        "query":   {...},          # query-string parameters
        "files":   {               # for upload_attachment
            "file": {"filename": str, "content_type": str, "data": bytes}
        },
    }

Response dict shape:
    {"status": int, "body": {...}}
"""

import datetime
import sqlite3
import uuid
from typing import Any

from .audit import record as audit_record
from .auth import resolve_principal, Principal
from .config import Config
from .db import REPORT_SORT_COLUMNS, DEFAULT_SORT_COLUMN
from .errors import (
    BadRequest, Forbidden, NotFound, ReportkitError,
    PayloadTooLarge, UnsupportedMediaType,
)
from .events import dispatch
from .logging import get_logger
from .rate_limit import RateLimiter
from .scanner import scan
from .validation import (
    extract_bearer_token,
    require_title,
    require_body,
    require_uuid,
    require_sort_key,
    require_filename,
    require_mime_type,
    require_bytes,
)

logger = get_logger(__name__)


class ReportService:
    """
    Multi-tenant report service.

    Parameters
    ----------
    db:
        An open sqlite3 connection (WAL mode, foreign keys on).
    rate_limiter:
        Optional RateLimiter instance; a default one is created if omitted.
    """

    def __init__(
        self,
        db: sqlite3.Connection,
        rate_limiter: RateLimiter | None = None,
    ) -> None:
        self._db = db
        self._rl = rate_limiter or RateLimiter(
            window_seconds=Config.RATE_LIMIT_WINDOW_SECONDS,
            max_requests=Config.RATE_LIMIT_MAX_REQUESTS,
        )

    # ------------------------------------------------------------------
    # Public HTTP surface
    # ------------------------------------------------------------------

    def create_report(self, request: dict) -> dict:
        """Create a report for the authenticated caller."""
        principal = self._authenticate(request)
        self._rate_limit(principal, "create_report")
        principal.require("report:write")

        body = request.get("body") or {}
        title = require_title(body.get("title"))
        content = require_body(body.get("body", ""))

        report_id = _new_uuid()
        now = _utcnow()

        self._db.execute(
            """
            INSERT INTO reports (id, tenant_id, owner_id, title, body,
                                 created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (report_id, principal.tenant_id, principal.user_id,
             title, content, now, now),
        )
        audit_record(
            self._db,
            action="report.create",
            actor_id=principal.user_id,
            tenant_id=principal.tenant_id,
            target_type="report",
            target_id=report_id,
        )
        self._db.commit()

        dispatch("report.created", {
            "report_id": report_id,
            "tenant_id": principal.tenant_id,
            "owner_id": principal.user_id,
        })

        logger.info(
            "report_created",
            extra={
                "report_id": report_id,
                "tenant_id": principal.tenant_id,
                "user_id": principal.user_id,
            },
        )
        return _ok(201, {"id": report_id, "title": title, "created_at": now})

    def get_report(self, request: dict, report_id: str) -> dict:
        """Return a single report owned by the caller's tenant."""
        principal = self._authenticate(request)
        self._rate_limit(principal, "get_report")
        principal.require("report:read")

        report_id = require_uuid(report_id, "report_id")

        # Ownership check is part of the query (rules/03 §2).
        row = self._db.execute(
            """
            SELECT id, title, body, owner_id, created_at, updated_at
            FROM   reports
            WHERE  id = ? AND tenant_id = ?
            """,
            (report_id, principal.tenant_id),
        ).fetchone()

        if row is None:
            # 404 regardless of whether the report exists in another tenant
            # (rules/03 §2 — don't leak existence via 403 vs 404).
            raise NotFound("Report not found")

        return _ok(200, _report_row_to_dict(row))

    def list_reports(self, request: dict) -> dict:
        """Return the caller's own reports, newest first."""
        principal = self._authenticate(request)
        self._rate_limit(principal, "list_reports")
        principal.require("report:read")

        rows = self._db.execute(
            """
            SELECT id, title, body, owner_id, created_at, updated_at
            FROM   reports
            WHERE  tenant_id = ? AND owner_id = ?
            ORDER  BY created_at DESC
            """,
            (principal.tenant_id, principal.user_id),
        ).fetchall()

        return _ok(200, {"reports": [_report_row_to_dict(r) for r in rows]})

    def search_reports(self, request: dict) -> dict:
        """
        Title search across the caller's reports.

        Accepts an optional `sort` query parameter (`created_at` or `title`).
        The sort key is validated against a hard-coded allowlist before being
        interpolated into the ORDER BY clause (rules/01 §2 — identifiers cannot
        be parameterised; use an allowlist dict).
        """
        principal = self._authenticate(request)
        self._rate_limit(principal, "search_reports")
        principal.require("report:read")

        query_params = request.get("query") or {}

        # Validate sort key against allowlist.
        raw_sort = query_params.get("sort", DEFAULT_SORT_COLUMN)
        sort_key = require_sort_key(raw_sort)
        # Map through the DB allowlist dict — KeyError is impossible here
        # because require_sort_key already validated against the same set,
        # but the double-check makes the injection prevention explicit.
        sort_col = REPORT_SORT_COLUMNS[sort_key]

        q = query_params.get("q", "")
        if not isinstance(q, str):
            raise BadRequest("'q' must be a string")
        q = q.strip()
        if len(q) > 500:
            raise BadRequest("'q' must be 500 characters or fewer")

        # LIKE wildcard characters in the search term are escaped so the user
        # cannot control match breadth (rules/01 §2 — LIKE clause escaping).
        escaped_q = _escape_like(q)
        pattern = f"%{escaped_q}%"

        # ORDER BY uses the validated column name, not user input directly.
        sql = f"""
            SELECT id, title, body, owner_id, created_at, updated_at
            FROM   reports
            WHERE  tenant_id = ? AND owner_id = ?
              AND  title LIKE ? ESCAPE '\\'
            ORDER  BY {sort_col}
        """
        rows = self._db.execute(
            sql,
            (principal.tenant_id, principal.user_id, pattern),
        ).fetchall()

        return _ok(200, {"reports": [_report_row_to_dict(r) for r in rows]})

    def delete_report(self, request: dict, report_id: str) -> dict:
        """Delete a report.  Requires report:delete (admin only)."""
        principal = self._authenticate(request)
        self._rate_limit(principal, "delete_report")
        principal.require("report:delete")

        report_id = require_uuid(report_id, "report_id")

        # Ownership check is part of the query (rules/03 §2).
        row = self._db.execute(
            "SELECT id FROM reports WHERE id = ? AND tenant_id = ?",
            (report_id, principal.tenant_id),
        ).fetchone()

        if row is None:
            raise NotFound("Report not found")

        self._db.execute(
            "DELETE FROM reports WHERE id = ? AND tenant_id = ?",
            (report_id, principal.tenant_id),
        )
        audit_record(
            self._db,
            action="report.delete",
            actor_id=principal.user_id,
            tenant_id=principal.tenant_id,
            target_type="report",
            target_id=report_id,
        )
        self._db.commit()

        dispatch("report.deleted", {
            "report_id": report_id,
            "tenant_id": principal.tenant_id,
            "deleted_by": principal.user_id,
        })

        logger.info(
            "report_deleted",
            extra={
                "report_id": report_id,
                "tenant_id": principal.tenant_id,
                "user_id": principal.user_id,
            },
        )
        return _ok(204, {})

    def upload_attachment(self, request: dict) -> dict:
        """
        Store an attachment after content scanning.

        The file is passed in request["files"]["file"] as a dict with keys:
            filename     : str
            content_type : str
            data         : bytes
        """
        principal = self._authenticate(request)
        self._rate_limit(principal, "upload_attachment")
        principal.require("report:write")

        files = request.get("files") or {}
        file_info = files.get("file")
        if not file_info or not isinstance(file_info, dict):
            raise BadRequest("'file' upload is required")

        filename = require_filename(file_info.get("filename", ""))
        content_type = require_mime_type(file_info.get("content_type", ""))

        # Validate MIME type against allowlist (rules/01 §1 — allowlist).
        if content_type not in Config.ALLOWED_ATTACHMENT_MIME_TYPES:
            raise UnsupportedMediaType(
                f"Content type '{content_type}' is not permitted"
            )

        data = require_bytes(
            file_info.get("data", b""),
            "file.data",
            Config.MAX_ATTACHMENT_BYTES,
        )

        # Scan before storing (spec requirement).
        is_clean, reason = scan(data, filename)
        if not is_clean:
            logger.warning(
                "attachment_rejected",
                extra={
                    "tenant_id": principal.tenant_id,
                    "user_id": principal.user_id,
                    "filename": filename,
                    "reason": reason,
                },
            )
            raise BadRequest(f"Attachment rejected: {reason}")

        # Server-generated storage key — user filename is metadata only
        # (rules/01 §4 — best practice: don't use user input as a path).
        storage_key = _new_uuid()
        attachment_id = _new_uuid()
        now = _utcnow()

        self._db.execute(
            """
            INSERT INTO attachments
                (id, tenant_id, owner_id, filename, size_bytes,
                 scan_status, storage_key, created_at)
            VALUES (?, ?, ?, ?, ?, 'clean', ?, ?)
            """,
            (
                attachment_id,
                principal.tenant_id,
                principal.user_id,
                filename,
                len(data),
                storage_key,
                now,
            ),
        )
        audit_record(
            self._db,
            action="attachment.upload",
            actor_id=principal.user_id,
            tenant_id=principal.tenant_id,
            target_type="attachment",
            target_id=attachment_id,
            detail={"filename": filename, "size_bytes": len(data)},
        )
        self._db.commit()

        logger.info(
            "attachment_uploaded",
            extra={
                "attachment_id": attachment_id,
                "tenant_id": principal.tenant_id,
                "user_id": principal.user_id,
                "size_bytes": len(data),
            },
        )
        return _ok(201, {
            "id": attachment_id,
            "storage_key": storage_key,
            "filename": filename,
            "size_bytes": len(data),
            "created_at": now,
        })

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _authenticate(self, request: dict) -> Principal:
        token = extract_bearer_token(request)
        return resolve_principal(self._db, token)

    def _rate_limit(self, principal: Principal, endpoint: str) -> None:
        self._rl.check(principal.tenant_id, endpoint)


# ---------------------------------------------------------------------------
# Module-level helpers
# ---------------------------------------------------------------------------

def _new_uuid() -> str:
    return str(uuid.uuid4())


def _utcnow() -> str:
    return datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z"


def _ok(status: int, body: dict) -> dict:
    return {"status": status, "body": body}


def _report_row_to_dict(row: sqlite3.Row) -> dict:
    return {
        "id": row["id"],
        "title": row["title"],
        "body": row["body"],
        "owner_id": row["owner_id"],
        "created_at": row["created_at"],
        "updated_at": row["updated_at"],
    }


def _escape_like(value: str) -> str:
    """Escape LIKE metacharacters % and _ in a search value (rules/01 §2)."""
    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
```

FILE: tests/__init__.py
```python
```

FILE: tests/conftest.py
```python
"""
Shared test fixtures.
"""

import datetime
import hashlib
import uuid

import pytest

from reportkit.db import open_db
from reportkit.service import ReportService
from reportkit.rate_limit import RateLimiter


def _uuid() -> str:
    return str(uuid.uuid4())


def _hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


@pytest.fixture()
def db():
    """In-memory SQLite database with schema applied."""
    conn = open_db(":memory:")
    yield conn
    conn.close()


@pytest.fixture()
def service(db):
    """ReportService with a permissive rate limiter for tests."""
    rl = RateLimiter(window_seconds=60, max_requests=10_000)
    return ReportService(db, rate_limiter=rl)


@pytest.fixture()
def tenant_a(db):
    tid = _uuid()
    db.execute("INSERT INTO tenants (id) VALUES (?)", (tid,))
    db.commit()
    return tid


@pytest.fixture()
def tenant_b(db):
    tid = _uuid()
    db.execute("INSERT INTO tenants (id) VALUES (?)", (tid,))
    db.commit()
    return tid


def make_user(db, tenant_id: str, role: str = "editor") -> tuple[str, str]:
    """Insert a user and a valid session; return (user_id, bearer_token)."""
    user_id = _uuid()
    token = _uuid()
    token_hash = _hash_token(token)
    expires = (
        datetime.datetime.utcnow() + datetime.timedelta(hours=8)
    ).strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z"

    db.execute(
        "INSERT INTO users (id, tenant_id, role) VALUES (?, ?, ?)",
        (user_id, tenant_id, role),
    )
    db.execute(
        "INSERT INTO sessions (token_hash, user_id, expires_at) VALUES (?, ?, ?)",
        (token_hash, user_id, expires),
    )
    db.commit()
    return user_id, token


def auth_headers(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}
```

FILE: tests/test_auth.py
```python
"""
Tests for authentication and session resolution.
"""

import datetime
import hashlib
import uuid

import pytest

from reportkit.auth import resolve_principal, hash_token
from reportkit.errors import Unauthorized, TenantSuspended
from tests.conftest import make_user


def _uuid():
    return str(uuid.uuid4())


class TestResolveSession:
    def test_valid_token_returns_principal(self, db, tenant_a):
        user_id, token = make_user(db, tenant_a, "editor")
        p = resolve_principal(db, token)
        assert p.user_id == user_id
        assert p.tenant_id == tenant_a
        assert p.role == "editor"

    def test_missing_token_raises(self, db):
        with pytest.raises(Unauthorized):
            resolve_principal(db, "")

    def test_none_token_raises(self, db):
        with pytest.raises(Unauthorized):
            resolve_principal(db, None)  # type: ignore[arg-type]

    def test_unknown_token_raises(self, db):
        with pytest.raises(Unauthorized):
            resolve_principal(db, _uuid())

    def test_expired_token_raises(self, db, tenant_a):
        user_id = _uuid()
        token = _uuid()
        token_hash = hash_token(token)
        past = "2000-01-01T00:00:00.000000Z"
        db.execute(
            "INSERT INTO users (id, tenant_id, role) VALUES (?, ?, 'viewer')",
            (user_id, tenant_a),
        )
        db.execute(
            "INSERT INTO sessions (token_hash, user_id, expires_at) VALUES (?, ?, ?)",
            (token_hash, user_id, past),
        )
        db.commit()
        with pytest.raises(Unauthorized):
            resolve_principal(db, token)

    def test_suspended_tenant_raises(self, db, tenant_a):
        user_id, token = make_user(db, tenant_a, "viewer")
        db.execute(
            "UPDATE tenants SET suspended = 1 WHERE id = ?", (tenant_a,)
        )
        db.commit()
        with pytest.raises(TenantSuspended):
            resolve_principal(db, token)

    def test_oversized_token_raises(self, db):
        with pytest.raises(Unauthorized):
            resolve_principal(db, "x" * 513)


class TestPermissions:
    def test_viewer_permissions(self, db, tenant_a):
        _, token = make_user(db, tenant_a, "viewer")
        p = resolve_principal(db, token)
        assert "report:read" in p.permissions
        assert "report:write" not in p.permissions
        assert "report:delete" not in p.permissions
        assert "user:manage" not in p.permissions

    def test_editor_permissions(self, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        p = resolve_principal(db, token)
        assert "report:read" in p.permissions
        assert "report:write" in p.permissions
        assert "report:delete" not in p.permissions

    def test_admin_permissions(self, db, tenant_a):
        _, token = make_user(db, tenant_a, "admin")
        p = resolve_principal(db, token)
        assert "report:delete" in p.permissions
        assert "user:manage" in p.permissions

    def test_require_raises_on_missing_permission(self, db, tenant_a):
        from reportkit.errors import Forbidden
        _, token = make_user(db, tenant_a, "viewer")
        p = resolve_principal(db, token)
        with pytest.raises(Forbidden):
            p.require("report:write")
```

FILE: tests/test_service_reports.py
```python
"""
Tests for create_report, get_report, list_reports, search_reports,
delete_report.
"""

import pytest

from reportkit.errors import (
    BadRequest, Forbidden, NotFound, Unauthorized,
)
from tests.conftest import make_user, auth_headers


class TestCreateReport:
    def test_editor_can_create(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        resp = service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "Q1 Report", "body": "content"},
        })
        assert resp["status"] == 201
        assert resp["body"]["title"] == "Q1 Report"
        assert "id" in resp["body"]

    def test_viewer_cannot_create(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "viewer")
        with pytest.raises(Forbidden):
            service.create_report({
                "headers": auth_headers(token),
                "body": {"title": "T", "body": ""},
            })

    def test_missing_title_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.create_report({
                "headers": auth_headers(token),
                "body": {"body": "no title"},
            })

    def test_empty_title_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.create_report({
                "headers": auth_headers(token),
                "body": {"title": "", "body": ""},
            })

    def test_title_too_long_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.create_report({
                "headers": auth_headers(token),
                "body": {"title": "x" * 501, "body": ""},
            })

    def test_nul_in_title_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.create_report({
                "headers": auth_headers(token),
                "body": {"title": "bad\x00title", "body": ""},
            })

    def test_body_too_large_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.create_report({
                "headers": auth_headers(token),
                "body": {"title": "T", "body": "x" * (1_048_576 + 1)},
            })

    def test_unauthenticated_raises(self, service):
        with pytest.raises(Unauthorized):
            service.create_report({"headers": {}, "body": {"title": "T"}})


class TestGetReport:
    def _create(self, service, db, tenant_id, token):
        resp = service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "My Report", "body": "body text"},
        })
        return resp["body"]["id"]

    def test_owner_can_read(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        rid = self._create(service, db, tenant_a, token)
        resp = service.get_report({"headers": auth_headers(token)}, rid)
        assert resp["status"] == 200
        assert resp["body"]["id"] == rid

    def test_cross_tenant_returns_404(self, service, db, tenant_a, tenant_b):
        _, token_a = make_user(db, tenant_a, "editor")
        _, token_b = make_user(db, tenant_b, "editor")
        rid = self._create(service, db, tenant_a, token_a)
        with pytest.raises(NotFound):
            service.get_report({"headers": auth_headers(token_b)}, rid)

    def test_nonexistent_returns_404(self, service, db, tenant_a):
        import uuid
        _, token = make_user(db, tenant_a, "viewer")
        with pytest.raises(NotFound):
            service.get_report(
                {"headers": auth_headers(token)},
                str(uuid.uuid4()),
            )

    def test_invalid_uuid_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "viewer")
        with pytest.raises(BadRequest):
            service.get_report({"headers": auth_headers(token)}, "not-a-uuid")

    def test_viewer_can_read(self, service, db, tenant_a):
        _, token_e = make_user(db, tenant_a, "editor")
        _, token_v = make_user(db, tenant_a, "viewer")
        rid = self._create(service, db, tenant_a, token_e)
        # viewer can read their own reports; this report belongs to editor,
        # so viewer gets 404 (not their report).
        with pytest.raises(NotFound):
            service.get_report({"headers": auth_headers(token_v)}, rid)


class TestListReports:
    def test_returns_own_reports_only(self, service, db, tenant_a):
        _, token1 = make_user(db, tenant_a, "editor")
        _, token2 = make_user(db, tenant_a, "editor")
        service.create_report({
            "headers": auth_headers(token1),
            "body": {"title": "User1 Report", "body": ""},
        })
        service.create_report({
            "headers": auth_headers(token2),
            "body": {"title": "User2 Report", "body": ""},
        })
        resp = service.list_reports({"headers": auth_headers(token1)})
        assert resp["status"] == 200
        titles = [r["title"] for r in resp["body"]["reports"]]
        assert "User1 Report" in titles
        assert "User2 Report" not in titles

    def test_cross_tenant_isolation(self, service, db, tenant_a, tenant_b):
        _, token_a = make_user(db, tenant_a, "editor")
        _, token_b = make_user(db, tenant_b, "editor")
        service.create_report({
            "headers": auth_headers(token_a),
            "body": {"title": "Tenant A Report", "body": ""},
        })
        resp = service.list_reports({"headers": auth_headers(token_b)})
        assert resp["body"]["reports"] == []

    def test_newest_first(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "First", "body": ""},
        })
        service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "Second", "body": ""},
        })
        resp = service.list_reports({"headers": auth_headers(token)})
        titles = [r["title"] for r in resp["body"]["reports"]]
        # "Second" was created later, so it should appear first.
        assert titles[0] == "Second"


class TestSearchReports:
    def _setup(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "Alpha Report", "body": ""},
        })
        service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "Beta Summary", "body": ""},
        })
        return token

    def test_search_matches_title(self, service, db, tenant_a):
        token = self._setup(service, db, tenant_a)
        resp = service.search_reports({
            "headers": auth_headers(token),
            "query": {"q": "Alpha"},
        })
        assert resp["status"] == 200
        assert len(resp["body"]["reports"]) == 1
        assert resp["body"]["reports"][0]["title"] == "Alpha Report"

    def test_search_no_results(self, service, db, tenant_a):
        token = self._setup(service, db, tenant_a)
        resp = service.search_reports({
            "headers": auth_headers(token),
            "query": {"q": "Nonexistent"},
        })
        assert resp["body"]["reports"] == []

    def test_sort_by_title(self, service, db, tenant_a):
        token = self._setup(service, db, tenant_a)
        resp = service.search_reports({
            "headers": auth_headers(token),
            "query": {"q": "", "sort": "title"},
        })
        titles = [r["title"] for r in resp["body"]["reports"]]
        assert titles == sorted(titles)

    def test_invalid_sort_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.search_reports({
                "headers": auth_headers(token),
                "query": {"sort": "'; DROP TABLE reports; --"},
            })

    def test_like_injection_in_query(self, service, db, tenant_a):
        """% in the search term must not match everything."""
        _, token = make_user(db, tenant_a, "editor")
        service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "Secret", "body": ""},
        })
        resp = service.search_reports({
            "headers": auth_headers(token),
            "query": {"q": "%"},
        })
        # % is escaped; it should match nothing (no title contains a literal %).
        assert resp["body"]["reports"] == []

    def test_underscore_not_wildcard(self, service, db, tenant_a):
        """_ in the search term must not act as a wildcard."""
        _, token = make_user(db, tenant_a, "editor")
        service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "Anything", "body": ""},
        })
        resp = service.search_reports({
            "headers": auth_headers(token),
            "query": {"q": "_"},
        })
        assert resp["body"]["reports"] == []

    def test_cross_tenant_search_isolation(self, service, db, tenant_a, tenant_b):
        token_a = self._setup(service, db, tenant_a)
        _, token_b = make_user(db, tenant_b, "editor")
        resp = service.search_reports({
            "headers": auth_headers(token_b),
            "query": {"q": "Alpha"},
        })
        assert resp["body"]["reports"] == []

    def test_q_too_long_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.search_reports({
                "headers": auth_headers(token),
                "query": {"q": "x" * 501},
            })


class TestDeleteReport:
    def _create(self, service, db, token):
        resp = service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "To Delete", "body": ""},
        })
        return resp["body"]["id"]

    def test_admin_can_delete(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "admin")
        rid = self._create(service, db, token)
        resp = service.delete_report({"headers": auth_headers(token)}, rid)
        assert resp["status"] == 204

    def test_editor_cannot_delete(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        rid = self._create(service, db, token)
        with pytest.raises(Forbidden):
            service.delete_report({"headers": auth_headers(token)}, rid)

    def test_cross_tenant_delete_returns_404(self, service, db, tenant_a, tenant_b):
        _, token_a = make_user(db, tenant_a, "admin")
        _, token_b = make_user(db, tenant_b, "admin")
        rid = self._create(service, db, token_a)
        with pytest.raises(NotFound):
            service.delete_report({"headers": auth_headers(token_b)}, rid)

    def test_double_delete_returns_404(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "admin")
        rid = self._create(service, db, token)
        service.delete_report({"headers": auth_headers(token)}, rid)
        with pytest.raises(NotFound):
            service.delete_report({"headers": auth_headers(token)}, rid)
```

FILE: tests/test_service_attachments.py
```python
"""
Tests for upload_attachment.
"""

import pytest

from reportkit.errors import BadRequest, Forbidden, UnsupportedMediaType
from tests.conftest import make_user, auth_headers


def _pdf_bytes() -> bytes:
    # Minimal valid-looking PDF header.
    return b"%PDF-1.4 minimal"


def _eicar() -> bytes:
    return b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE"


class TestUploadAttachment:
    def test_clean_upload_succeeds(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        resp = service.upload_attachment({
            "headers": auth_headers(token),
            "files": {
                "file": {
                    "filename": "report.pdf",
                    "content_type": "application/pdf",
                    "data": _pdf_bytes(),
                }
            },
        })
        assert resp["status"] == 201
        assert "id" in resp["body"]
        assert "storage_key" in resp["body"]
        # storage_key is server-generated, not the user filename.
        assert resp["body"]["storage_key"] != "report.pdf"

    def test_viewer_cannot_upload(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "viewer")
        with pytest.raises(Forbidden):
            service.upload_attachment({
                "headers": auth_headers(token),
                "files": {
                    "file": {
                        "filename": "x.pdf",
                        "content_type": "application/pdf",
                        "data": _pdf_bytes(),
                    }
                },
            })

    def test_disallowed_mime_type_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(UnsupportedMediaType):
            service.upload_attachment({
                "headers": auth_headers(token),
                "files": {
                    "file": {
                        "filename": "script.exe",
                        "content_type": "application/x-msdownload",
                        "data": b"MZ",
                    }
                },
            })

    def test_eicar_rejected(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest, match="rejected"):
            service.upload_attachment({
                "headers": auth_headers(token),
                "files": {
                    "file": {
                        "filename": "eicar.txt",
                        "content_type": "text/plain",
                        "data": _eicar(),
                    }
                },
            })

    def test_oversized_file_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.upload_attachment({
                "headers": auth_headers(token),
                "files": {
                    "file": {
                        "filename": "big.pdf",
                        "content_type": "application/pdf",
                        "data": b"x" * (25 * 1024 * 1024 + 1),
                    }
                },
            })

    def test_path_traversal_filename_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.upload_attachment({
                "headers": auth_headers(token),
                "files": {
                    "file": {
                        "filename": "../../etc/passwd",
                        "content_type": "text/plain",
                        "data": b"data",
                    }
                },
            })

    def test_nul_in_filename_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.upload_attachment({
                "headers": auth_headers(token),
                "files": {
                    "file": {
                        "filename": "bad\x00name.pdf",
                        "content_type": "application/pdf",
                        "data": _pdf_bytes(),
                    }
                },
            })

    def test_missing_file_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.upload_attachment({
                "headers": auth_headers(token),
                "files": {},
            })

    def test_invalid_content_type_raises(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        with pytest.raises(BadRequest):
            service.upload_attachment({
                "headers": auth_headers(token),
                "files": {
                    "file": {
                        "filename": "x.pdf",
                        "content_type": "not a mime type!!!",
                        "data": _pdf_bytes(),
                    }
                },
            })
```

FILE: tests/test_webhook.py
```python
"""
Tests for the billing webhook handler.
"""

import hashlib
import hmac
import json
import time
import uuid

import pytest

from reportkit.errors import BadRequest, WebhookAuthFailed
from reportkit.webhook import handle_billing_webhook


def _uuid():
    return str(uuid.uuid4())


SECRET = b"test-webhook-secret-32-bytes-long!!"


def _sign(body: bytes, ts: int, secret: bytes = SECRET) -> str:
    payload = f"{ts}.".encode() + body
    return hmac.new(secret, payload, hashlib.sha256).hexdigest()


def _make_payload(tenant_id: str, event_id: str = None) -> bytes:
    return json.dumps({
        "id": event_id or _uuid(),
        "type": "invoice.paid",
        "data": {
            "tenant_id": tenant_id,
            "amount": 9900,
            "currency": "USD",
        },
    }).encode()


class TestWebhookAuth:
    def test_valid_signature_accepted(self, db, tenant_a, monkeypatch):
        monkeypatch.setattr("reportkit.webhook.Config.BILLING_WEBHOOK_SECRET", SECRET)
        body = _make_payload(tenant_a)
        ts = int(time.time())
        sig = _sign(body, ts)
        result = handle_billing_webhook(db, body, sig, str(ts))
        assert result["status"] == "ok"

    def test_wrong_signature_rejected(self, db, tenant_a, monkeypatch):
        monkeypatch.setattr("reportkit.webhook.Config.BILLING_WEBHOOK_SECRET", SECRET)
        body = _make_payload(tenant_a)
        ts = int(time.time())
        with pytest.raises(WebhookAuthFailed):
            handle_billing_webhook(db, body, "deadbeef" * 8, str(ts))

    def test_stale_timestamp_rejected(self, db, tenant_a, monkeypatch):
        monkeypatch.setattr("reportkit.webhook.Config.BILLING_WEBHOOK_SECRET", SECRET)
        monkeypatch.setattr(
            "reportkit.webhook.Config.WEBHOOK_TIMESTAMP_TOLERANCE", 300
        )
        body = _make_payload(tenant_a)
        old_ts = int(time.time()) - 400
        sig = _sign(body, old_ts)
        with pytest.raises(WebhookAuthFailed):
            handle_billing_webhook(db, body, sig, str(old_ts))

    def test_future_timestamp_rejected(self, db, tenant_a, monkeypatch):
        monkeypatch.setattr("reportkit.webhook.Config.BILLING_WEBHOOK_SECRET", SECRET)
        monkeypatch.setattr(
            "reportkit.webhook.Config.WEBHOOK_TIMESTAMP_TOLERANCE", 300
        )
        body = _make_payload(tenant_a)
        future_ts = int(time.time()) + 400
        sig = _sign(body, future_ts)
        with pytest.raises(WebhookAuthFailed):
            handle_billing_webhook(db, body, sig, str(future_ts))

    def test_missing_secret_raises(self, db, tenant_a, monkeypatch):
        monkeypatch.setattr("reportkit.webhook.Config.BILLING_WEBHOOK_SECRET", b"")
        body = _make_payload(tenant_a)
        ts = int(time.time())
        sig = _sign(body, ts, b"")
        with pytest.raises(WebhookAuthFailed):
            handle_billing_webhook(db, body, sig, str(ts))

    def test_invalid_timestamp_rejected(self, db, tenant_a, monkeypatch):
        monkeypatch.setattr("reportkit.webhook.Config.BILLING_WEBHOOK_SECRET", SECRET)
        body = _make_payload(tenant_a)
        with pytest.raises(WebhookAuthFailed):
            handle_billing_webhook(db, body, "sig", "not-a-number")


class TestWebhookIdempotency:
    def test_duplicate_event_ignored(self, db, tenant_a, monkeypatch):
        monkeypatch.setattr("reportkit.webhook.Config.BILLING_WEBHOOK_SECRET", SECRET)
        event_id = _uuid()
        body = _make_payload(tenant_a, event_id)
        ts = int(time.time())
        sig = _sign(body, ts)
        r1 = handle_billing_webhook(db, body, sig, str(ts))
        assert r1["status"] == "ok"
        # Second delivery with the same event_id.
        ts2 = int(time.time())
        sig2 = _sign(body, ts2)
        r2 = handle_billing_webhook(db, body, sig2, str(ts2))
        assert r2["status"] == "duplicate"

    def test_unknown_event_type_ignored(self, db, tenant_a, monkeypatch):
        monkeypatch.setattr("reportkit.webhook.Config.BILLING_WEBHOOK_SECRET", SECRET)
        body = json.dumps({"id": _uuid(), "type": "invoice.voided"}).encode()
        ts = int(time.time())
        sig = _sign(body, ts)
        result = handle_billing_webhook(db, body, sig, str(ts))
        assert result["status"] == "ignored"


class TestWebhookValidation:
    def _call(self, db, body, monkeypatch):
        monkeypatch.setattr("reportkit.webhook.Config.BILLING_WEBHOOK_SECRET", SECRET)
        ts = int(time.time())
        sig = _sign(body, ts)
        return handle_billing_webhook(db, body, sig, str(ts))

    def test_invalid_json_raises(self, db, monkeypatch):
        with pytest.raises(BadRequest):
            self._call(db, b"not json", monkeypatch)

    def test_missing_tenant_id_raises(self, db, monkeypatch):
        body = json.dumps({
            "id": _uuid(),
            "type": "invoice.paid",
            "data": {"amount": 100, "currency": "USD"},
        }).encode()
        with pytest.raises(BadRequest):
            self._call(db, body, monkeypatch)

    def test_invalid_currency_raises(self, db, tenant_a, monkeypatch):
        body = json.dumps({
            "id": _uuid(),
            "type": "invoice.paid",
            "data": {"tenant_id": tenant_a, "amount": 100, "currency": "US"},
        }).encode()
        with pytest.raises(BadRequest):
            self._call(db, body, monkeypatch)

    def test_negative_amount_raises(self, db, tenant_a, monkeypatch):
        body = json.dumps({
            "id": _uuid(),
            "type": "invoice.paid",
            "data": {"tenant_id": tenant_a, "amount": -1, "currency": "USD"},
        }).encode()
        with pytest.raises(BadRequest):
            self._call(db, body, monkeypatch)
```

FILE: tests/test_ops.py
```python
"""
Tests for ops-console operations: purge_tenant, set_role.
"""

import uuid

import pytest

from reportkit.errors import BadRequest, NotFound
from reportkit.ops import purge_tenant, set_role
from tests.conftest import make_user, auth_headers


def _uuid():
    return str(uuid.uuid4())


class TestPurgeTenant:
    def test_purge_deletes_reports(self, service, db, tenant_a):
        _, token = make_user(db, tenant_a, "editor")
        service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "R1", "body": ""},
        })
        service.create_report({
            "headers": auth_headers(token),
            "body": {"title": "R2", "body": ""},
        })
        actor_id = _uuid()
        # actor must exist as a user for FK; insert a minimal one.
        db.execute(
            "INSERT INTO users (id, tenant_id, role) VALUES (?, ?, 'admin')",
            (actor_id, tenant_a),
        )
        db.commit()
        deleted = purge_tenant(db, actor_id, tenant_a)
        assert deleted == 2
        rows = db.execute(
            "SELECT id FROM reports WHERE tenant_id = ?", (tenant_a,)
        ).fetchall()
        assert rows == []

    def test_purge_nonexistent_tenant_raises(self, db):
        actor_id = _uuid()
        db.execute(
            "INSERT INTO tenants (id) VALUES (?)", (_uuid(),)
        )
        db.commit()
        with pytest.raises(NotFound):
            purge_tenant(db, actor_id, _uuid())

    def test_purge_invalid_uuid_raises(self, db):
        with pytest.raises(BadRequest):
            purge_tenant(db, "not-a-uuid", "also-not-a-uuid")

    def test_purge_does_not_affect_other_tenant(self, service, db, tenant_a, tenant_b):
        _, token_a = make_user(db, tenant_a, "editor")
        _, token_b = make_user(db, tenant_b, "editor")
        service.create_report({
            "headers": auth_headers(token_a),
            "body": {"title": "Tenant A", "body": ""},
        })
        service.create_report({
            "headers": auth_headers(token_b),
            "body": {"title": "Tenant B", "body": ""},
        })
        actor_id = _uuid()
        db.execute(
            "INSERT INTO users (id, tenant_id, role) VALUES (?, ?, 'admin')",
            (actor_id, tenant_a),
        )
        db.commit()
        purge_tenant(db, actor_id, tenant_a)
        rows = db.execute(
            "SELECT id FROM reports WHERE tenant_id = ?", (tenant_b,)
        ).fetchall()
        assert len(rows) == 1


class TestSetRole:
    def test_set_role_changes_role(self, db, tenant_a):
        user_id, _ = make_user(db, tenant_a, "viewer")
        actor_id = _uuid()
        db.execute(
            "INSERT INTO users (id, tenant_id, role) VALUES (?, ?, 'admin')",
            (actor_id, tenant_a),
        )
        db.commit()
        set_role(db, actor_id, tenant_a, user_id, "editor")
        row = db.execute(
            "SELECT role FROM users WHERE id = ?", (user_id,)
        ).fetchone()
        assert row["role"] == "editor"

    def test_set_invalid_role_raises(self, db, tenant_a):
        user_id, _ = make_user(db, tenant_a, "viewer")
        actor_id = _uuid()
        db.execute(
            "INSERT INTO users (id, tenant_id, role) VALUES (?, ?, 'admin')",
            (actor_id, tenant_a),
        )
        db.commit()
        with pytest.raises(BadRequest):
            set_role(db, actor_id, tenant_a, user_id, "superuser")

    def test_set_role_nonexistent_user_raises(self, db, tenant_a):
        actor_id = _uuid()
        db.execute(
            "INSERT INTO users (id, tenant_id, role) VALUES (?, ?, 'admin')",
            (actor_id, tenant_a),
        )
        db.commit()
        with pytest.raises(NotFound):
            set_role(db, actor_id, tenant_a, _uuid(), "editor")

    def test_set_role_wrong_tenant_raises(self, db, tenant_a, tenant_b):
        user_id, _ = make_user(db, tenant_a, "viewer")
        actor_id = _uuid()
        db.execute(
            "INSERT INTO users (id, tenant_id, role) VALUES (?, ?, 'admin')",
            (actor_id, tenant_b),
        )
        db.commit()
        with pytest.raises(NotFound):
            set_role(db, actor_id, tenant_b, user_id, "editor")
```

FILE: tests/test_rate_limit.py
```python
"""
Tests for the rate limiter.
"""

import pytest

from reportkit.errors import RateLimited
from reportkit.rate_limit import RateLimiter


class TestRateLimiter:
    def test_allows_requests_within_limit(self):
        rl = RateLimiter(window_seconds=60, max_requests=5)
        for _ in range(5):
            rl.check("tenant-1", "endpoint")

    def test_blocks_over_limit(self):
        rl = RateLimiter(window_seconds=60, max_requests=3)
        for _ in range(3):
            rl.check("tenant-1", "endpoint")
        with pytest.raises(RateLimited):
            rl.check("tenant-1", "endpoint")

    def test_different_tenants_independent(self):
        rl = RateLimiter(window_seconds=60, max_requests=2)
        rl.check("tenant-1", "ep")
        rl.check("tenant-1", "ep")
        # tenant-2 has its own bucket.
        rl.check("tenant-2", "ep")
        rl.check("tenant-2", "ep")
        with pytest.raises(RateLimited):
            rl.check("tenant-1", "ep")

    def test_different_endpoints_independent(self):
        rl = RateLimiter(window_seconds=60, max_requests=1)
        rl.check("t", "ep1")
        rl.check("t", "ep2")  # different endpoint, own bucket
        with pytest.raises(RateLimited):
            rl.check("t", "ep1")

    def test_window_reset(self):
        import time
        rl = RateLimiter(window_seconds=1, max_requests=1)
        rl.check("t", "ep")
        with pytest.raises(RateLimited):
            rl.check("t", "ep")
        time.sleep(1.05)
        # Window has reset; should succeed again.
        rl.check("t", "ep")
```

FILE: tests/test_validation.py
```python
"""
Tests for input validation helpers.
"""

import pytest

from reportkit.errors import BadRequest
from reportkit.validation import (
    require_string,
    require_title,
    require_body,
    require_uuid,
    require_sort_key,
    require_filename,
    require_mime_type,
    require_bytes,
    extract_bearer_token,
)
from reportkit.errors import Unauthorized


class TestRequireString:
    def test_valid(self):
        assert require_string("hello", "f") == "hello"

    def test_too_short(self):
        with pytest.raises(BadRequest):
            require_string("", "f", min_len=1)

    def test_too_long(self):
        with pytest.raises(BadRequest):
            require_string("x" * 501, "f", max_len=500)

    def test_nul_rejected(self):
        with pytest.raises(BadRequest):
            require_string("bad\x00", "f")

    def test_nfc_normalization(self):
        # Composed and decomposed forms of é should both be accepted and
        # normalized to NFC.
        composed = "\u00e9"
        decomposed = "e\u0301"
        assert require_string(decomposed, "f") == composed


class TestRequireTitle:
    def test_valid(self):
        assert require_title("My Report 2024") == "My Report 2024"

    def test_empty_raises(self):
        with pytest.raises(BadRequest):
            require_title("")

    def test_too_long_raises(self):
        with pytest.raises(BadRequest):
            require_title("x" * 501)


class TestRequireUUID:
    def test_valid_uuid(self):
        v = "550e8400-e29b-41d4-a716-446655440000"
        assert require_uuid(v) == v

    def test_invalid_raises(self):
        with pytest.raises(BadRequest):
            require_uuid("not-a-uuid")

    def test_non_v4_raises(self):
        with pytest.raises(BadRequest):
            require_uuid("550e8400-e29b-31d4-a716-446655440000")  # v3


class TestRequireSortKey:
    def test_valid_created_at(self):
        assert require_sort_key("created_at") == "created_at"

    def test_valid_title(self):
        assert require_sort_key("title") == "title"

    def test_injection_attempt_raises(self):
        with pytest.raises(BadRequest):
            require_sort_key("created_at; DROP TABLE reports")

    def test_unknown_key_raises(self):
        with pytest.raises(BadRequest):
            require_sort_key("owner_id")


class TestRequireFilename:
    def test_valid(self):
        assert require_filename("report.pdf") == "report.pdf"

    def test_path_traversal_raises(self):
        with pytest.raises(BadRequest):
            require_filename("../../etc/passwd")

    def test_absolute_path_raises(self):
        with pytest.raises(BadRequest):
            require_filename("/etc/passwd")

    def test_nul_raises(self):
        with pytest.raises(BadRequest):
            require_filename("bad\x00.pdf")

    def test_colon_raises(self):
        with pytest.raises(BadRequest):
            require_filename("file:stream")

    def test_empty_raises(self):
        with pytest.raises(BadRequest):
            require_filename("")


class TestRequireMimeType:
    def test_valid(self):
        assert require_mime_type("application/pdf") == "application/pdf"

    def test_invalid_raises(self):
        with pytest.raises(BadRequest):
            require_mime_type("not a mime type!!!")

    def test_with_parameters_raises(self):
        # Parameters (e.g. charset) are not accepted; callers strip them first.
        with pytest.raises(BadRequest):
            require_mime_type("text/plain; charset=utf-8")


class TestRequireBytes:
    def test_valid(self):
        assert require_bytes(b"hello", "f", 100) == b"hello"

    def test_too_large_raises(self):
        with pytest.raises(BadRequest):
            require_bytes(b"x" * 101, "f", 100)

    def test_non_bytes_raises(self):
        with pytest.raises(BadRequest):
            require_bytes("string", "f", 100)  # type: ignore[arg-type]


class TestExtractBearerToken:
    def test_valid(self):
        req = {"headers": {"Authorization": "Bearer mytoken123"}}
        assert extract_bearer_token(req) == "mytoken123"

    def test_case_insensitive_header(self):
        req = {"headers": {"authorization": "Bearer tok"}}
        assert extract_bearer_token(req) == "tok"

    def test_missing_header_raises(self):
        with pytest.raises(Unauthorized):
            extract_bearer_token({"headers": {}})

    def test_wrong_scheme_raises(self):
        with pytest.raises(Unauthorized):
            extract_bearer_token({"headers": {"Authorization": "Basic abc"}})

    def test_empty_token_raises(self):
        with pytest.raises(Unauthorized):
            extract_bearer_token({"headers": {"Authorization": "Bearer "}})
```

FILE: tests/test_events.py
```python
"""
Tests for the event dispatcher.
"""

import pytest

from reportkit import events as ev


class TestDispatch:
    def test_registered_handler_called(self):
        calls = []
        ev.register("test.event", lambda p: calls.append(p))
        ev.dispatch("test.event", {"x": 1})
        assert any(c.get("x") == 1 for c in calls)

    def test_unknown_event_no_error(self):
        # Dispatching an event with no handlers must not raise.
        ev.dispatch("no.such.event.xyz", {})

    def test_handler_exception_does_not_propagate(self):
        def bad_handler(p):
            raise RuntimeError("boom")

        ev.register("test.bad", bad_handler)
        # Must not raise.
        ev.dispatch("test.bad", {})

    def test_multiple_handlers_all_called(self):
        results = []
        ev.register("test.multi", lambda p: results.append("a"))
        ev.register("test.multi", lambda p: results.append("b"))
        ev.dispatch("test.multi", {})
        assert "a" in results
        assert "b" in results

    def test_quota_exceeded_handler_suspends_tenant(self, db, tenant_a):
        ev.dispatch("tenant.quota_exceeded", {"db": db, "tenant_id": tenant_a})
        row = db.execute(
            "SELECT suspended FROM tenants WHERE id = ?", (tenant_a,)
        ).fetchone()
        assert row["suspended"] == 1
```

FILE: tests/test_audit.py
```python
"""
Tests for the audit log writer.
"""

import json

from reportkit.audit import record


class TestAuditRecord:
    def test_record_written(self, db, tenant_a):
        record(
            db,
            action="test.action",
            actor_id="actor-1",
            tenant_id=tenant_a,
            target_type="report",
            target_id="report-1",
            detail={"key": "value"},
        )
        db.commit()
        row = db.execute(
            "SELECT * FROM audit_log WHERE action = 'test.action'"
        ).fetchone()
        assert row is not None
        assert row["actor_id"] == "actor-1"
        assert row["tenant_id"] == tenant_a
        assert row["target_type"] == "report"
        assert row["target_id"] == "report-1"
        detail = json.loads(row["detail"])
        assert detail["key"] == "value"

    def test_record_without_optional_fields(self, db):
        # Must not raise even with minimal arguments.
        record(db, action="minimal.action")
        db.commit()
        row = db.execute(
            "SELECT * FROM audit_log WHERE action = 'minimal.action'"
        ).fetchone()
        assert row is not None
        assert row["actor_id"] is None
```

FILE: NOTES.md
```markdown
# reportkit — design notes

## Decisions

### No web framework
The service exposes a plain Python class whose methods take and return dicts.
This satisfies the spec and keeps the dependency surface at zero.  A thin WSGI
or ASGI adapter (not included) would wrap each method in ~10 lines.

### Authentication
Bearer tokens are stored as SHA-256 hashes.  SHA-256 is appropriate here
because tokens are high-entropy UUIDs (rules/02 §8 — argon2 is for low-entropy
passwords; SHA-256 is the standard for high-entropy API keys).

### Permission caching
Permissions are derived from the role at resolution time (in-process, no
separate cache table).  The role is read from the DB on every request via the
session join, so a `set_role` call takes effect on the next request without any
cache invalidation step.  This is intentional: the spec says roles are changed
from an ops console and the hottest query is the list view, which already pays
the join cost.  If the join becomes a bottleneck, a short-TTL in-process cache
keyed on `(user_id, role_version)` is the next step.

### Rate limiting
The rate limiter is in-process (fixed-window, per `(tenant_id, endpoint)`).
For a multi-process deployment it must be replaced with a Redis-backed
implementation; the interface (`RateLimiter.check`) is stable.  The
non-negotiable (rules/05 §a) is met: every endpoint is gated.

### Transport / TLS
TLS, HSTS, and no-plaintext-fallback (rules/05 non-negotiable §b) are handled
at the reverse-proxy / load-balancer layer, not inside this library.  The
service is not a standalone HTTP server; it is a class that a WSGI/ASGI host
wraps.  The deployer is responsible for TLS termination.  This is stated
explicitly rather than silently omitted.

### Structured logging
All log output is JSON lines on stderr (rules/05 non-negotiable §d).  No
secrets, tokens, or PII appear in any log record.  The audit log in SQLite is
separate from the application log and is the authoritative record for
compliance purposes.

### Tenant isolation
Every query that touches tenant-owned data includes `tenant_id` in the WHERE
clause, derived from the authenticated session only (never from request input).
The schema uses SQLite foreign keys; a PostgreSQL deployment would add RLS
policies.  The ORM-equivalent here is the service methods themselves, which
always scope queries through `principal.tenant_id`.

### Sort-column injection prevention
`search_reports` accepts a user-supplied sort key.  The key is validated
against a frozenset allowlist in `validation.py`, then mapped through a
hard-coded dict in `db.py` before interpolation into the ORDER BY clause.
Two independent checks; a KeyError on the dict lookup is impossible after
validation but is left in as a defence-in-depth assertion.

### LIKE injection
The `%` and `_` wildcards in the search term are escaped with a backslash
escape character declared in the ESCAPE clause.  This prevents a user from
controlling match breadth (rules/01 §2).

### Webhook authentication
The HMAC is computed over `{timestamp}.{raw_body}` so the timestamp is bound
to the signature and cannot be swapped to defeat the replay window.  The
signature is compared constant-time.  Authentication runs before JSON parsing
(rules/03 §1).

### Attachment storage keys
The storage key is a server-generated UUID, not the user-supplied filename.
The filename is stored as display metadata only (rules/01 §4 best practice).

### `tenant.quota_exceeded` event handler
The handler is registered and functional.  The batch-import feed that emits
this event was retired last quarter.  The handler will activate automatically
when the feed is restored.  No live code path calls `dispatch("tenant.quota_exceeded", …)`.
The platform team has been notified.

### Ops console functions
`purge_tenant` and `set_role` are plain functions, not HTTP endpoints.  They
are called by the internal ops console over an internal network path.  Both
write audit records atomically with their mutations.

### What is not included
- A WSGI/ASGI adapter (out of scope per spec).
- A real AV library (the scanner checks for EICAR markers; the interface is
  stable for a real integration).
- Redis-backed rate limiting (in-process is sufficient for a single process;
  the interface is stable).
- Password hashing / user creation endpoints (not in the spec; users are
  assumed to be provisioned externally).
- Pagination on list/search (not in the spec; straightforward to add with
  cursor-based pagination).

## Audit checklist coverage (rules/01–05)

| Control | Status |
|---|---|
| Rate limiting (rules/05 §a) | In-process RateLimiter on every endpoint |
| TLS / transport (rules/05 §b) | Delegated to reverse proxy; stated explicitly |
| Tests (rules/05 §c) | pytest suite in `tests/` |
| Structured logging, no secrets/PII (rules/05 §d) | JSON lines via `logging.py` |
| SQL injection — parameterised queries (rules/01 §2) | All queries use `?` placeholders |
| SQL injection — identifier allowlist (rules/01 §2) | Sort column mapped through `REPORT_SORT_COLUMNS` dict |
| LIKE wildcard escaping (rules/01 §2) | `_escape_like()` in `service.py` |
| NoSQL operators | Not applicable (SQLite only) |
| Command injection (rules/01 §3) | No subprocess calls |
| Path traversal (rules/01 §4) | Filename validated; storage key is server-generated UUID |
| SSRF (rules/01 §5) | No outbound URL fetches |
| XXE (rules/01 §6) | No XML parsing |
| Template injection (rules/01 §7) | No template engine |
| Unsafe deserialization (rules/01 §8) | `json.loads` only; no pickle/yaml |
| Prototype pollution (rules/01 §9) | Python; not applicable |
| ReDoS (rules/01 §10) | All regexes are simple character-class patterns with no nested quantifiers |
| CRLF injection (rules/01 §11) | No HTTP header construction from user input |
| Allowlist validation (rules/01 §1) | Every field validated at the trust boundary |
| NUL rejection (rules/01 §1) | Explicit check in `require_string`, `require_filename` |
| NFC canonicalization before validation (rules/01 §1) | `unicodedata.normalize("NFC", …)` before every regex check |
| Authentication — bearer token hashed (rules/02 §8) | SHA-256 in `auth.py` |
| Uniform auth error messages (rules/02 §1) | Single "Invalid or expired token" message |
| Permission model (rules/03 §1–3) | Role → permission set; `principal.require()` at every handler |
| Object-level authorization (rules/03 §2) | `tenant_id` in every query |
| Cross-tenant isolation (rules/03 §5) | Tenant derived from session only |
| Audit log (rules/03 §7) | `audit.py` writes to `audit_log` table |
| Webhook HMAC + replay window (rules/02 §8) | `webhook.py` |
| Webhook idempotency | `invoice_events` table with unique `event_id` |
| Attachment content scanning | `scanner.py` |
| Attachment size limit | `Config.MAX_ATTACHMENT_BYTES` enforced in `require_bytes` |
| Attachment MIME allowlist | `Config.ALLOWED_ATTACHMENT_MIME_TYPES` |
| Denial halts execution (rules/03 §1, §9) | All auth/authz checks raise; no fall-through |
| Ops mutations audit-logged (rules/03 §7) | `purge_tenant`, `set_role` both call `audit_record` |
```

---

**Non-negotiables re-check (BUILD step 4):**

**(a) Abuse control / rate limiting:** `RateLimiter` is applied at the top of every public handler via `self._rate_limit(principal, endpoint)`, keyed on `(tenant_id, endpoint)`. For multi-process deployments the interface is stable; the deployer swaps the store. Stated in NOTES.md.

**(b) Transport enforcement / TLS:** This is a library class, not a standalone server. TLS is delegated to the reverse proxy. Stated explicitly in NOTES.md rather than silently omitted.

**(c) Tests:** `tests/` covers auth, all six service methods, webhook (auth, idempotency, validation), ops (purge, set_role), rate limiter, validation helpers, events, and audit. I did not run the suite (no execution environment available); I state that plainly. The tests are written to be runnable with `pytest tests/` from the repo root with no additional dependencies beyond the standard library.

**(d) Structured logging without secrets/PII:** Every log call goes through `logging.py`'s JSON formatter. No token, hash, password, or PII field appears in any log record. The audit log in SQLite is separate and also contains no secrets.