Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions examples/serve_with_password.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
import sys
from textual_serve.server import Server

if __name__ == "__main__":
command = sys.argv[1] if len(sys.argv) > 1 else "python -m textual"
password = sys.argv[2] if len(sys.argv) > 2 else None
server = Server(command, password=password)
server.serve()
114 changes: 113 additions & 1 deletion src/textual_serve/server.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,13 @@
from __future__ import annotations

import asyncio
import hashlib
import hmac

import logging
import os
import secrets
import time
from pathlib import Path
import signal
import sys
Expand Down Expand Up @@ -36,6 +40,31 @@

WINDOWS = sys.platform == "WINDOWS"

AUTH_COOKIE_NAME = "textual_serve_auth"
AUTH_COOKIE_MAX_AGE = 86400 * 30 # 30 days


def _make_secret() -> str:
"""Generate a random secret for HMAC signing."""
return secrets.token_hex(32)


def _sign_cookie(secret: str, payload: str) -> str:
"""Sign a payload with HMAC-SHA256. Returns 'payload.hexdigest'."""
mac = hmac.new(secret.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256)
return f"{payload}.{mac.hexdigest()}"


def _verify_cookie(secret: str, signed: str) -> str | None:
"""Verify a signed cookie value. Returns payload if valid, None otherwise."""
if "." not in signed:
return None
payload, _, sig = signed.rpartition(".")
expected = _sign_cookie(secret, payload)
if hmac.compare_digest(expected, signed):
return payload
return None


class LogHighlighter(RegexHighlighter):
base_style = "repr."
Expand Down Expand Up @@ -74,6 +103,7 @@ def __init__(
public_url: str | None = None,
statics_path: str | os.PathLike = "./static",
templates_path: str | os.PathLike = "./templates",
password: str | None = None,
):
"""

Expand Down Expand Up @@ -105,6 +135,10 @@ def __init__(
self.templates_path = base_path / templates_path
self.console = Console()
self.download_manager = DownloadManager()
self._password = password
self._auth_secret: str | None = None
if password is not None:
self._auth_secret = _make_secret()

def initialize_logging(self) -> None:
"""Initialize logging.
Expand Down Expand Up @@ -132,13 +166,86 @@ def request_exit(self) -> None:
"""Gracefully exit the app."""
raise GracefulExit()

def _check_auth(self, request: web.Request) -> bool:
"""Return True if the request has a valid auth cookie."""
if self._password is None or self._auth_secret is None:
return True
signed = request.cookies.get(AUTH_COOKIE_NAME, "")
payload = _verify_cookie(self._auth_secret, signed)
if payload is None:
return False
try:
ts = int(payload)
if time.time() - ts > AUTH_COOKIE_MAX_AGE:
return False
except (ValueError, TypeError):
return False
return True

def _set_auth_cookie(self, response: web.Response) -> None:
"""Set the auth cookie on a response."""
if self._auth_secret is None:
return
payload = str(int(time.time()))
signed = _sign_cookie(self._auth_secret, payload)
response.set_cookie(
AUTH_COOKIE_NAME,
signed,
max_age=AUTH_COOKIE_MAX_AGE,
httponly=True,
samesite="Lax",
)

def _clear_auth_cookie(self, response: web.Response) -> None:
"""Clear the auth cookie."""
response.del_cookie(AUTH_COOKIE_NAME)

@web.middleware
async def _auth_middleware(
self, request: web.Request, handler: callable
) -> web.StreamResponse:
"""Middleware that enforces password authentication."""
# Static files are always accessible (needed for login page assets)
if request.path.startswith("/static/"):
return await handler(request)

# No password configured — everything is open
if self._password is None:
return await handler(request)

# Check existing cookie
authenticated = self._check_auth(request)

# Query-param bootstrap: /?password=correct sets the cookie
query_pass = request.query.get("password", "")
if query_pass:
if query_pass == self._password:
resp = web.HTTPFound("/")
self._set_auth_cookie(resp)
return resp
else:
request["auth_error"] = "Incorrect password"
authenticated = False

# Protected API routes
if not authenticated:
if request.path == "/ws":
raise web.HTTPForbidden(
text="Authentication required. Access the main page first."
)
if request.path.startswith("/download/"):
raise web.HTTPForbidden(text="Authentication required.")

request["authenticated"] = authenticated
return await handler(request)

async def _make_app(self) -> web.Application:
"""Make the aiohttp web.Application.

Returns:
New aiohttp web application.
"""
app = web.Application()
app = web.Application(middlewares=[self._auth_middleware])

aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader(self.templates_path))

Expand Down Expand Up @@ -248,6 +355,9 @@ async def handle_index(self, request: web.Request) -> dict[str, Any]:
router = request.app.router
font_size = to_int(request.query.get("fontsize", "16"), 16)

authenticated = request.get("authenticated", True)
auth_error = request.get("auth_error", "")

def get_url(route: str, **args) -> str:
"""Get a URL from the aiohttp router."""
path = router[route].url_for(**args)
Expand All @@ -265,6 +375,8 @@ def get_websocket_url(route: str, **args) -> str:
context = {
"font_size": font_size,
"app_websocket_url": get_websocket_url("websocket"),
"authenticated": authenticated,
"auth_error": auth_error,
}
context["config"] = {
"static": {
Expand Down
81 changes: 81 additions & 0 deletions src/textual_serve/templates/app_index.html
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,50 @@
#start.-delay {
display: flex;
}

.login-brand {
display: flex;
flex-direction: column;
align-items: center;
}

.login-brand svg {
padding-right: 0;
margin-bottom: 8px;
}

.login-separator {
width: 1px;
height: 120px;
background: rgba(255, 255, 255, 0.15);
margin: 0 32px;
}

.login-form {
display: flex;
flex-direction: column;
align-items: center;
gap: 12px;
}

.login-form input[type="password"] {
padding: 8px 12px;
font-family: "Roboto Mono", menlo, monospace;
font-size: 14px;
background: #1a3a4a;
border: 1px solid #5e0ba7;
color: rgba(255, 255, 255, 0.9);
outline: none;
}

.login-form input[type="password"]:focus {
border-color: #ac5af4;
}

.auth-error {
color: #e74c3c;
margin-top: 8px;
}
</style>
<script>
function getStartUrl() {
Expand All @@ -120,6 +164,42 @@
</script>
</head>
<body data-pingurl="{{ ping_url }}">
{% if not authenticated %}
<div class="dialog-container intro-dialog">
<div class="shade"></div>
<div class="intro">
<div class="login-brand">
<svg
width="32px"
viewBox="0 0 2933 2261"
fill="none"
xmlns="http://www.w3.org/2000/svg"
>
<path
fill-rule="evenodd"
clip-rule="evenodd"
d="M2927.81 0H1677.1L1312.35 205.173H306.689L0.359375 434.921L204.029 893.177H735.972L644.784 2261H1060.36L1441.72 1974.97L2073.92 688.003H2334.4L2717.9 472.286L2927.81 0ZM1245.82 410.347L1276.32 277.656H330.85L153.929 410.347H1245.82ZM1229.16 482.83H100.972L251.134 820.694H813.449L722.26 2188.52H837.043L1229.16 482.83ZM1350.7 277.656L911.417 2188.52H1023.87L1662.19 615.52H2301.36L2451.52 277.656H1350.7ZM1460.19 205.173H2497.79L2733.69 72.4829H1696.09L1460.19 205.173Z"
fill="#ffffff"
/>
</svg>
<div>{{ application.name or 'Textual Application' }}</div>
{% if auth_error %}
<div class="auth-error">{{ auth_error }}</div>
{% endif %}
</div>
<div class="login-separator"></div>
<form class="login-form" method="GET" action="/">
<input
type="password"
name="password"
placeholder="Password"
autofocus
/>
<button type="submit">Login</button>
</form>
</div>
</div>
{% else %}
<div class="dialog-container intro-dialog">
<div class="shade"></div>
<div class="intro">
Expand Down Expand Up @@ -156,5 +236,6 @@
data-session-websocket-url="{{ app_websocket_url }}"
data-font-size="{{ font_size }}"
></div>
{% endif %}
</body>
</html>