feat(audit): optional best-effort webhook for clipboard and file events

docs/settings.md

@@ -47,6 +47,9 @@ Every setting below is one entry of `SETTING_DEFINITIONS` in [`src/selkies/setti
 | `--webrtc-pacer` | `SELKIES_WEBRTC_PACER` | bool | `true` | Pace outgoing WebRTC packets per transport with strict priorities (audio/RTCP > data-channel > video), an IDR-aware video queue budget and GOP-reset recovery, so audio and interactive signaling are protected from video bursts on congested links. Enabled by default; set SELKIES_WEBRTC_PACER=false to disable. SELKIES_WEBRTC_PACER_STALE_MS sets the stale-GOP purge deadline in milliseconds (0 = disabled). |
 | `--file-transfers` | `SELKIES_FILE_TRANSFERS` | list: any of `upload`, `download` | `upload,download` | Allowed file transfer directions (comma-separated: "upload,download"). Set to "" or "none" to disable. |
 | `--file-transfer-limit-mbps` | `SELKIES_FILE_TRANSFER_LIMIT_MBPS` | float, from 0.0 | `0.0` | Static file-transfer throttle in Mbit/s, one allowance shared by all downloads and uploads, for links whose rate the operator knows. 0 disables. The congestion-control pacing protects the video stream without it, in both directions and end to end through a reverse proxy; the cap is for holding transfers to a fixed share regardless. |
+| `--audit-webhook-url` | `SELKIES_AUDIT_WEBHOOK_URL` | str | (empty) | URL that receives one JSON POST per clipboard transfer, file upload and file download, carrying metadata only (the event, an RFC 3339 timestamp, byte size, MIME type or file name) and never the content. Events are delivered in order over one keep-alive connection; a collector that is slow or down loses what overflows the queue rather than stalling the session. Empty (default) sends nothing. |
+| `--audit-webhook-token` | `SELKIES_AUDIT_WEBHOOK_TOKEN` | str | (empty) | Bearer token sent in the Authorization header of every audit POST. Empty sends no header. Never sent to clients. |
+| `--audit-webhook-timeout` | `SELKIES_AUDIT_WEBHOOK_TIMEOUT` | float, from 0.1 | `2.0` | Seconds one audit POST may take before it counts as failed and the next event is sent. |
 | `--framerate` | `SELKIES_FRAMERATE` | range, within `8-240` | `60` | Framerate: allowed range (e.g., "8-240"), initial value (e.g., "60"), or both ("60,8-240"); "60-60" locks. |
 | `--video-crf` | `SELKIES_VIDEO_CRF` | range, within `5-50` | `25` | Video CRF (constant quality): allowed range (e.g., "5-50"), initial value (e.g., "25"), or both ("25,5-50"); "25-25" locks. |
 | `--video-bitrate` | `SELKIES_VIDEO_BITRATE` | range, within `100-1000000` | `8000` | Video bitrate aka CBR, in kilobits per second (kbps): allowed range (e.g., "100-1000000"), initial value (e.g., "8000" for 8 Mbps, "250" for 250 kbps), or both ("8000,100-1000000"); "8000-8000" locks. |

docs/usage.md

@@ -44,6 +44,20 @@ The side menu's files section uploads files into the session and browses the sam
 
 `--file-manager-path` (`FILE_MANAGER_PATH`, default `~/Desktop`) is the directory both directions use, created at startup when missing. Transfers in either direction are paced against the video stream so a large one does not stall the session, measured end to end so a reverse proxy in front changes nothing; `--file-transfer-limit-mbps` adds a fixed cap on top for operators who want one.
 
+## Audit Trail
+
+`--audit-webhook-url` (`SELKIES_AUDIT_WEBHOOK_URL`) POSTs one JSON object to a collector for every clipboard transfer, file upload and file download, for deployments that have to produce a record of what moved. Metadata only: the content is never sent, and neither is anything identifying the client, since Selkies has no first-class user of its own. Nothing is sent without a URL. `--audit-webhook-token` adds an `Authorization: Bearer` header a proxy in front of the collector can check, and `--audit-webhook-timeout` bounds one POST; the URL is used as given, so a collector anywhere but this host wants `https://`.
+
+| `event` | Fields | Recorded when |
+| --- | --- | --- |
+| `clipboard.send` | `mime_type`, `size_bytes` | the session's clipboard goes out to the clients |
+| `clipboard.receive` | `mime_type`, `size_bytes`, `multipart` | a client's clipboard is written into the session |
+| `file.upload.end` | `filename`, `size_bytes` | an upload lands in the file-manager directory |
+| `file.upload.error` | `filename`, `error` | an upload is refused or fails, `filename` as the client asked for it |
+| `file.download` | `filename`, `size_bytes` | a file is served out of the file-manager directory |
+
+Every object also carries `ts`, an RFC 3339 UTC timestamp with milliseconds taken when the transfer happened, and both transports emit the same objects. An event costs the session an enqueue and nothing else: one task delivers the queue in order over a single keep-alive connection, so a collector that is slow or down never paces the stream. The queue holds 1024 events and drops what overflows, a POST that fails drops its event with no retry, and each outage is logged once.
+
 ## Session Sharing
 
 The side menu's sharing section hands out links to the running session. Each one is the page's own address with a fragment on the end, and it carries no credential of its own — whatever already guards the page (HTTP Basic authentication, a reverse proxy) guards the link too, so treat a copied link as one:

src/selkies/__main__.py

@@ -31,6 +31,7 @@
 from .selkies import DataStreamingServer
 from .stream_server import CentralizedStreamServer
 from .webcam import stop_shared_webcam
+from . import audit
 
 
 logging.basicConfig(level=logging.INFO)
@@ -123,6 +124,7 @@ async def run() -> None:
         await server.run()
     finally:
         await stop_shared_webcam()
+        await audit.close()
 
 
 def main() -> None:

src/selkies/audit.py

@@ -0,0 +1,100 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+"""Audit trail of clipboard and file transfers, POSTed to an operator's webhook.
+
+Every transfer the server carries — clipboard content in either direction,
+a file upload, a file download — is one JSON object on `audit_webhook_url`:
+its `event`, an RFC 3339 `ts`, and metadata (byte size, MIME type or file
+name), never the content. Events queue in order and one task delivers them
+over a single keep-alive connection, so a transfer pays an enqueue and
+nothing else; a collector that is slow or down loses what overflows the
+queue rather than stalling a session, and each outage is logged once.
+Without a URL every call is a no-op.
+"""
+
+import asyncio
+import logging
+import time
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+import aiohttp
+
+from .settings import settings
+
+logger = logging.getLogger("audit")
+
+QUEUE_BOUND = 1024
+
+_queue: Optional[asyncio.Queue] = None
+_sender: Optional[asyncio.Task] = None
+_overflowing = False
+_closing = False
+
+
+def emit(event: str, **fields: Any) -> None:
+    """Queue one event and return at once."""
+    global _queue, _sender, _overflowing
+    if not settings.audit_webhook_url:
+        return
+    if _queue is None:
+        _queue = asyncio.Queue(QUEUE_BOUND)
+        _sender = asyncio.get_running_loop().create_task(_deliver(_queue))
+    try:
+        _queue.put_nowait({"event": event, "ts": time.time(), **fields})
+        _overflowing = False
+    except asyncio.QueueFull:
+        if not _overflowing:
+            logger.warning("Audit webhook queue full (%d events); dropping events until it drains", QUEUE_BOUND)
+            _overflowing = True
+
+
+async def _deliver(queue: asyncio.Queue) -> None:
+    """Send queued events one by one; an outage is logged once, its end too."""
+    headers = {}
+    if settings.audit_webhook_token:
+        headers["Authorization"] = f"Bearer {settings.audit_webhook_token}"
+    timeout = aiohttp.ClientTimeout(total=settings.audit_webhook_timeout)
+    failing = False
+    async with aiohttp.ClientSession(headers=headers, timeout=timeout,
+                                     connector=aiohttp.TCPConnector(limit=1)) as session:
+        while not _closing:
+            payload = await queue.get()
+            payload["ts"] = datetime.fromtimestamp(payload["ts"], timezone.utc).isoformat(
+                timespec="milliseconds").replace("+00:00", "Z")
+            try:
+                async with session.post(settings.audit_webhook_url, json=payload) as response:
+                    failure = f"HTTP {response.status}" if response.status >= 400 else ""
+            except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc:
+                failure = str(exc) or type(exc).__name__
+            finally:
+                queue.task_done()
+            if failure and not failing:
+                logger.warning("Audit webhook %s failed: %s; events are dropped until it answers",
+                               settings.audit_webhook_url, failure)
+            elif failing and not failure:
+                logger.info("Audit webhook delivering again")
+            failing = bool(failure)
+
+
+async def close() -> None:
+    """Deliver what is queued, within one request timeout, and stop the sender."""
+    global _queue, _sender, _closing
+    if _sender is None:
+        return
+    # The flag ends the loop at an event boundary, so shutting down never means
+    # interrupting a request in flight; the cancel only breaks the idle wait.
+    _closing = True
+    try:
+        await asyncio.wait_for(_queue.join(), settings.audit_webhook_timeout)
+    except asyncio.TimeoutError:
+        pass
+    _sender.cancel()
+    try:
+        await _sender
+    except asyncio.CancelledError:
+        pass
+    _queue = _sender = None
+    _closing = False

src/selkies/input_handler.py

@@ -94,6 +94,7 @@
 )
 from .media_pipeline import RateControlMode
 from .settings import settings
+from . import audit
 try:
     from pixelflux import VirtualKeyboardUnavailable as PixelfluxVkUnavailable
 except Exception:
@@ -7729,6 +7730,7 @@ async def _dispatch_message(self, msg: str, display_id: str = 'primary',
                     # must find the clipboard set. Bytes pass straight through; a
                     # multi-MB decode and re-encode on the loop would be redundant.
                     if await self.write_clipboard(data, mime_type=mime_type):
+                        audit.emit("clipboard.receive", mime_type=mime_type, size_bytes=len(data), multipart=True)
                         if mime_type == "text/plain":
                             logger_webrtc_input.info(f"Set multi-part clipboard content, length: {len(data)}")
                         else:
@@ -7819,6 +7821,7 @@ async def _send_requested_clipboard():
                     data_bytes = base64.b64decode(b64_data)
                     # In-line so a paste keystroke right behind it pastes this content.
                     if await self.write_clipboard(data_bytes, mime_type=mime_type):
+                        audit.emit("clipboard.receive", mime_type=mime_type, size_bytes=len(data_bytes), multipart=False)
                         logger_webrtc_input.info(f"Set binary clipboard content ({mime_type}), size: {len(data_bytes)} bytes")
                 except Exception as e:
                     logger_webrtc_input.error(f"Binary clipboard write error: {e}")
@@ -7830,6 +7833,7 @@ async def _send_requested_clipboard():
                     data = base64.b64decode(toks[1]).decode("utf-8", 'ignore')
                     # In-line for paste-after-copy ordering (see the cb branch).
                     if await self.write_clipboard(data):
+                        audit.emit("clipboard.receive", mime_type="text/plain", size_bytes=len(data.encode()), multipart=False)
                         logger_webrtc_input.info(f"Set clipboard content, length: {len(data)}")
                 except Exception as e:
                     logger_webrtc_input.error(f"Clipboard decode error: {e}")

src/selkies/rtc.py

@@ -63,6 +63,7 @@
     pcmflux = None
 
 from .settings import settings as app_settings, inflate_gz_bounded, pipeline_starts_on, software_encoders, software_video_path
+from . import audit
 from .ice import TcpMux, UdpMux
 from .ice.ice import get_host_addresses
 from .webcam import CODEC_BY_NAME, get_shared_webcam, webcam_locked_off, webcam_uplink_allowed
@@ -643,6 +644,9 @@ async def send_clipboard_data(self, data: Union[str, bytes], mime_type: str = "t
             if channel is None or channel.readyState != "open":
                 return
             requester = channel
+        if data_bytes and (requester is not None
+                           or next(self._iter_open_data_channels(), None) is not None):
+            audit.emit("clipboard.send", mime_type=mime_type, size_bytes=len(data_bytes))
 
         def send_typed(msg_type: str, payload: Any) -> None:
             if requester is not None:

src/selkies/selkies.py

@@ -104,6 +104,7 @@
 )
 from .settings import settings, CODEC_LABELS, SETTING_DEFINITIONS, WS_MAX_MESSAGE_BYTES, WS_MESSAGE_SIZE_HARD_CAP, build_client_settings_payload, codec_for_encoder, effective_use_cpu, encoder_for_codec, inflate_gz_bounded, pipeline_starts_on, sanitize_client_setting
 from .settings import settings as app_settings
+from . import audit
 from .webcam import (
     MSG_WEBCAM_DISABLED,
     MSG_WEBCAM_KEYFRAME,
@@ -898,6 +899,8 @@ async def send_ws_clipboard_data(
                 return
             data_bytes = data.encode('utf-8') if not is_binary and isinstance(data, str) else data
             total_size = len(data_bytes)
+            if total_size:
+                audit.emit("clipboard.send", mime_type=mime_type, size_bytes=total_size)
             clients = self.data_streaming_server.clients
             # One payload at a time per client: the start/data/finish frames
             # carry no transfer id, so a send racing another (a clipboard

src/selkies/settings.py

@@ -203,6 +203,25 @@ def inflate_gz_bounded(payload: bytes) -> str:
         "min": 0.0,
         "help": 'Static file-transfer throttle in Mbit/s, one allowance shared by all downloads and uploads, for links whose rate the operator knows. 0 disables. The congestion-control pacing protects the video stream without it, in both directions and end to end through a reverse proxy; the cap is for holding transfers to a fixed share regardless.',
     },
+    {
+        "name": "audit_webhook_url",
+        "type": "str",
+        "default": "",
+        "help": 'URL that receives one JSON POST per clipboard transfer, file upload and file download, carrying metadata only (the event, an RFC 3339 timestamp, byte size, MIME type or file name) and never the content. Events are delivered in order over one keep-alive connection; a collector that is slow or down loses what overflows the queue rather than stalling the session. Empty (default) sends nothing.',
+    },
+    {
+        "name": "audit_webhook_token",
+        "type": "str",
+        "default": "",
+        "help": 'Bearer token sent in the Authorization header of every audit POST. Empty sends no header.',
+    },
+    {
+        "name": "audit_webhook_timeout",
+        "type": "float",
+        "default": 2.0,
+        "min": 0.1,
+        "help": 'Seconds one audit POST may take before it counts as failed and the next event is sent.',
+    },
     {
         "name": "framerate",
         "type": "range",
@@ -1075,6 +1094,7 @@ def inflate_gz_bounded(payload: bytes) -> str:
     "turn_password",
     "cloudflare_turn_token_id",
     "cloudflare_turn_api_token",
+    "audit_webhook_token",
 })
 for _setting_def in SETTING_DEFINITIONS:
     if _setting_def["name"] in SENSITIVE_SETTING_NAMES:
@@ -1844,6 +1864,7 @@ def effective_use_cpu(encoder: str, requested: Optional[bool], default: bool) ->
     'webcam_socket_path', 'webcam_device',
     'uinput_mouse_socket', 'webrtc_statistics_dir', 'computer_use_bind',
     'wayland_host_display', 'app_wayland_display',
+    'audit_webhook_url', 'audit_webhook_timeout',
 ]