Static HTTP File Server (python, written by Codex)

Codex wrote this python project from the task below. It does not run on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Static HTTP File Server

Write a program that serves static files over HTTP from a local directory, supporting directory listings, MIME type detection, caching headers, range requests for large files, and access logging.

FUNCTIONAL REQUIREMENTS:
- Accept a directory path to serve as a command-line argument (default: current directory)
- Start an HTTP server on a configurable port via --port flag (default: 8080) and bind address via --host flag (default: 0.0.0.0)
- Serve files with correct MIME types auto-detected from file extensions (HTML, CSS, JS, images, fonts, JSON, XML, PDF, video, audio, etc.)
- Generate directory listing pages when accessing a directory URL, showing file names, sizes, modification dates, and file type icons, with a --no-listing flag to disable this
- Support index files: automatically serve index.html or index.htm if present in a directory
- Implement HTTP caching headers: ETag based on file content hash, Last-Modified from file timestamp, and Cache-Control with configurable max-age via --cache flag (default: 3600 seconds)
- Support HTTP Range requests for partial content delivery (needed for video/audio streaming and resumable downloads)
- Log all requests to console in Common Log Format (CLF): remote address, timestamp, method, path, status code, response size, and response time
- Save access logs to a file via --log flag
- Support CORS headers via --cors flag to enable cross-origin requests (configurable allowed origins)
- Support HTTPS via --ssl flag with --cert and --key flags for certificate and private key file paths
- Print server startup information: URL, served directory, features enabled
- If no directory is given, create a sample directory with HTML, CSS, JS, image, and text files, then start serving them and print the URL to access each
- Handle errors: port already in use, permission denied, symlink traversal prevention, and graceful shutdown on SIGINT

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# Static HTTP File Server (Python)

Serves static files over HTTP/HTTPS with MIME detection, directory listing, cache headers, range requests, CORS, and CLF access logging.

## Requirements
- Ubuntu 22.04
- Python 3.10+

## Dependencies
- No external dependencies (Python standard library only)
- `requirements.txt` is intentionally empty of third-party packages

## Setup
```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

## Run
Serve a directory:
```bash
python3 src/main.py /path/to/serve
```

Custom host/port/cache:
```bash
python3 src/main.py . --host 127.0.0.1 --port 9090 --cache 120
```

Disable listing and enable CORS:
```bash
python3 src/main.py . --no-listing --cors "*"
```

Access log file:
```bash
python3 src/main.py . --log access.log
```

HTTPS:
```bash
python3 src/main.py . --ssl --cert cert.pem --key key.pem
```

Demo mode (no directory argument):
```bash
python3 src/main.py
```

## Features
- Correct MIME type serving
- Directory listings with name/size/modified/type indicator
- `index.html` / `index.htm` support
- `ETag`, `Last-Modified`, and `Cache-Control`
- HTTP range (`Range: bytes=...`) support
- CLF-style logging to console and optional file
- CORS via `--cors`
- HTTPS via `--ssl --cert --key`
- Symlink traversal prevention
- Graceful `SIGINT` shutdown

## Example startup output
```text
Static HTTP File Server started
URL: http://0.0.0.0:8080
Serving directory: /home/user/project
Directory listing: enabled
Caching max-age: 3600
CORS: disabled
TLS: disabled
```

# No external dependencies required.
# This project uses only Python 3.10+ standard library modules.

#!/usr/bin/env python3
"""Static HTTP file server with directory listing, caching, ranges, logging, CORS, and HTTPS."""

from __future__ import annotations

import argparse
import datetime as dt
import email.utils
import hashlib
import html
import mimetypes
import os
import signal
import ssl
import sys
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Optional, TextIO
from urllib.parse import unquote, urlsplit


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Static HTTP File Server")
    parser.add_argument("directory", nargs="?", help="Directory to serve")
    parser.add_argument("--host", default="0.0.0.0", help="Bind host (default: 0.0.0.0)")
    parser.add_argument("--port", type=int, default=8080, help="Port (default: 8080)")
    parser.add_argument("--no-listing", action="store_true", help="Disable directory listing")
    parser.add_argument("--cache", type=int, default=3600, help="Cache-Control max-age seconds")
    parser.add_argument("--log", dest="log_path", help="Access log file path")
    parser.add_argument("--cors", help="Allowed origin for CORS, e.g. '*' or https://site.example")
    parser.add_argument("--ssl", action="store_true", help="Enable HTTPS")
    parser.add_argument("--cert", help="Certificate file path for HTTPS")
    parser.add_argument("--key", help="Private key file path for HTTPS")
    args = parser.parse_args()

    if not (1 <= args.port <= 65535):
        parser.error("--port must be 1..65535")
    if args.cache < 0:
        parser.error("--cache must be >= 0")
    if args.ssl and (not args.cert or not args.key):
        parser.error("--ssl requires --cert and --key")
    return args


def ensure_sample_directory(root: Path) -> None:
    root.mkdir(parents=True, exist_ok=True)
    (root / "assets").mkdir(exist_ok=True)
    (root / "scripts").mkdir(exist_ok=True)

    (root / "index.html").write_text(
        """<!doctype html>
<html>
  <head>
    <meta charset="utf-8" />
    <title>Sample Static Server</title>
    <link rel="stylesheet" href="/assets/style.css" />
  </head>
  <body>
    <h1>Sample Static Site</h1>
    <p>Static file server is running.</p>
    <img src="/sample.svg" width="180" alt="sample image" />
    <p><a href="/hello.txt">Open sample text file</a></p>
    <script src="/scripts/app.js"></script>
  </body>
</html>
""",
        encoding="utf-8",
    )
    (root / "assets" / "style.css").write_text(
        "body { font-family: sans-serif; margin: 2rem; } h1 { color: #1a5d8f; }\n", encoding="utf-8"
    )
    (root / "scripts" / "app.js").write_text("console.log('Sample JavaScript loaded');\n", encoding="utf-8")
    (root / "hello.txt").write_text("Hello from sample static directory.\n", encoding="utf-8")
    (root / "sample.svg").write_text(
        """<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 320 120">
  <rect width="320" height="120" fill="#e9f3fb"/>
  <circle cx="60" cy="60" r="32" fill="#1a5d8f"/>
  <text x="110" y="68" font-size="22" fill="#1a5d8f">Static Server</text>
</svg>
""",
        encoding="utf-8",
    )


def human_size(size: int) -> str:
    units = ["B", "KB", "MB", "GB"]
    value = float(size)
    idx = 0
    while value >= 1024 and idx < len(units) - 1:
        value /= 1024.0
        idx += 1
    return f"{value:.0f} {units[idx]}" if idx == 0 else f"{value:.1f} {units[idx]}"


def clf_timestamp(now: dt.datetime) -> str:
    return now.strftime("%d/%b/%Y:%H:%M:%S %z")


def etag_for_file(file_path: Path) -> str:
    sha = hashlib.sha256()
    with file_path.open("rb") as fh:
        while True:
            chunk = fh.read(1024 * 1024)
            if not chunk:
                break
            sha.update(chunk)
    return f'"{sha.hexdigest()}"'


def icon_for(name: str, is_dir: bool) -> str:
    if is_dir:
        return "[DIR]"
    ext = Path(name).suffix.lower()
    if ext in {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"}:
        return "[IMG]"
    if ext in {".mp4", ".webm", ".mp3", ".wav", ".ogg"}:
        return "[MED]"
    if ext in {".html", ".htm", ".css", ".js", ".json", ".xml", ".txt", ".md"}:
        return "[TXT]"
    return "[FIL]"


class StaticHandler(BaseHTTPRequestHandler):
    server_version = "StaticHTTPFileServer/1.0"

    def do_OPTIONS(self) -> None:  # noqa: N802
        self.send_response(204)
        self._add_cors_headers()
        self.end_headers()
        self._write_access_log(204, 0, 0.0)

    def do_GET(self) -> None:  # noqa: N802
        self._handle_method(send_body=True)

    def do_HEAD(self) -> None:  # noqa: N802
        self._handle_method(send_body=False)

    def log_message(self, fmt: str, *args) -> None:  # noqa: A003
        return

    @property
    def root_real(self) -> Path:
        return self.server.root_real  # type: ignore[attr-defined]

    def _handle_method(self, send_body: bool) -> None:
        start = time.perf_counter()
        status = 500
        size_sent = 0
        try:
            if self.command not in {"GET", "HEAD"}:
                self._send_text(405, "Method Not Allowed")
                status = 405
                size_sent = 23 if send_body else 0
                return

            req_path = urlsplit(self.path).path
            target = self._resolve_path(req_path)
            if not target.exists():
                self._send_text(404, "Not Found")
                status = 404
                size_sent = 14 if send_body else 0
                return

            if target.is_dir():
                index_html = target / "index.html"
                index_htm = target / "index.htm"
                if index_html.is_file():
                    target = index_html
                elif index_htm.is_file():
                    target = index_htm
                else:
                    if self.server.no_listing:  # type: ignore[attr-defined]
                        self._send_text(403, "Directory Listing Disabled")
                        status = 403
                        size_sent = 31 if send_body else 0
                        return
                    status, size_sent = self._send_directory_listing(req_path, target, send_body)
                    return

            if not target.is_file():
                self._send_text(403, "Forbidden")
                status = 403
                size_sent = 14 if send_body else 0
                return

            status, size_sent = self._send_file(target, send_body)
        except PermissionError:
            self._send_text(403, "Forbidden")
            status = 403
            size_sent = 14 if send_body else 0
        except Exception:
            self._send_text(500, "Internal Server Error")
            status = 500
            size_sent = 26 if send_body else 0
        finally:
            elapsed_ms = (time.perf_counter() - start) * 1000.0
            self._write_access_log(status, size_sent, elapsed_ms)

    def _resolve_path(self, req_path: str) -> Path:
        decoded = unquote(req_path)
        joined = self.root_real / decoded.lstrip("/")
        resolved = Path(os.path.realpath(joined))
        root_str = str(self.root_real)
        resolved_str = str(resolved)
        if resolved_str != root_str and not resolved_str.startswith(root_str + os.sep):
            raise PermissionError("Traversal blocked")
        return resolved

    def _send_text(self, code: int, text: str) -> None:
        body = f"{code} {text}\n".encode("utf-8")
        self.send_response(code)
        self._add_cors_headers()
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(body)

    def _send_directory_listing(self, req_path: str, directory: Path, send_body: bool) -> tuple[int, int]:
        entries = sorted(directory.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower()))
        path_display = req_path if req_path.endswith("/") else req_path + "/"
        rows = []
        if path_display != "/":
            parent = str(Path(path_display.rstrip("/")).parent).replace("\\", "/")
            if not parent.endswith("/"):
                parent += "/"
            rows.append('<tr><td>[UP]</td><td><a href="%s">..</a></td><td>-</td><td>-</td></tr>' % html.escape(parent))
        for entry in entries:
            stat = entry.stat()
            name = entry.name + ("/" if entry.is_dir() else "")
            href = (path_display.rstrip("/") + "/" + entry.name).replace("//", "/")
            if entry.is_dir():
                href += "/"
            rows.append(
                "<tr>"
                f"<td>{icon_for(entry.name, entry.is_dir())}</td>"
                f'<td><a href="{html.escape(href)}">{html.escape(name)}</a></td>'
                f"<td>{'-' if entry.is_dir() else human_size(stat.st_size)}</td>"
                f"<td>{dt.datetime.fromtimestamp(stat.st_mtime).isoformat(sep=' ', timespec='seconds')}</td>"
                "</tr>"
            )
        html_body = (
            "<!doctype html><html><head><meta charset='utf-8'>"
            f"<title>Index of {html.escape(path_display)}</title>"
            "<style>body{font-family:sans-serif;margin:1.2rem}table{border-collapse:collapse;width:100%}"
            "td,th{border-bottom:1px solid #ddd;padding:0.5rem;text-align:left}</style></head><body>"
            f"<h1>Index of {html.escape(path_display)}</h1>"
            "<table><thead><tr><th>Type</th><th>Name</th><th>Size</th><th>Modified</th></tr></thead>"
            f"<tbody>{''.join(rows)}</tbody></table></body></html>"
        ).encode("utf-8")
        self.send_response(200)
        self._add_cors_headers()
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(html_body)))
        self.end_headers()
        if send_body:
            self.wfile.write(html_body)
        return 200, len(html_body) if send_body else 0

    def _parse_range(self, range_header: str, size: int) -> Optional[tuple[int, int]]:
        if not range_header.startswith("bytes="):
            return None
        spec = range_header[len("bytes=") :]
        if "," in spec:
            return None
        start_s, _, end_s = spec.partition("-")
        try:
            if start_s == "" and end_s == "":
                return None
            if start_s == "":
                suffix_len = int(end_s)
                if suffix_len <= 0:
                    return None
                start = max(0, size - suffix_len)
                end = size - 1
            else:
                start = int(start_s)
                end = int(end_s) if end_s else size - 1
            if start < 0 or end < start or start >= size:
                return None
            if end >= size:
                end = size - 1
            return start, end
        except ValueError:
            return None

    def _send_file(self, file_path: Path, send_body: bool) -> tuple[int, int]:
        stat = file_path.stat()
        size = stat.st_size
        etag = etag_for_file(file_path)
        last_modified = email.utils.formatdate(stat.st_mtime, usegmt=True)

        if_none_match = self.headers.get("If-None-Match")
        if_modified_since = self.headers.get("If-Modified-Since")
        if if_none_match == etag:
            self.send_response(304)
            self._add_cors_headers()
            self.send_header("ETag", etag)
            self.send_header("Last-Modified", last_modified)
            self.send_header("Cache-Control", f"public, max-age={self.server.cache_max_age}")  # type: ignore[attr-defined]
            self.end_headers()
            return 304, 0
        if if_modified_since:
            try:
                ims_dt = email.utils.parsedate_to_datetime(if_modified_since)
                file_dt = dt.datetime.fromtimestamp(stat.st_mtime, tz=dt.timezone.utc)
                if ims_dt and ims_dt.tzinfo and ims_dt >= file_dt:
                    self.send_response(304)
                    self._add_cors_headers()
                    self.send_header("ETag", etag)
                    self.send_header("Last-Modified", last_modified)
                    self.send_header("Cache-Control", f"public, max-age={self.server.cache_max_age}")  # type: ignore[attr-defined]
                    self.end_headers()
                    return 304, 0
            except Exception:
                pass

        range_header = self.headers.get("Range")
        start = 0
        end = size - 1
        status = 200
        if range_header:
            parsed = self._parse_range(range_header, size)
            if not parsed:
                self.send_response(416)
                self._add_cors_headers()
                self.send_header("Content-Range", f"bytes */{size}")
                self.end_headers()
                return 416, 0
            start, end = parsed
            status = 206

        ctype, _ = mimetypes.guess_type(str(file_path))
        if not ctype:
            ctype = "application/octet-stream"

        content_len = end - start + 1
        self.send_response(status)
        self._add_cors_headers()
        self.send_header("Content-Type", ctype)
        self.send_header("Accept-Ranges", "bytes")
        self.send_header("ETag", etag)
        self.send_header("Last-Modified", last_modified)
        self.send_header("Cache-Control", f"public, max-age={self.server.cache_max_age}")  # type: ignore[attr-defined]
        self.send_header("Content-Length", str(content_len))
        if status == 206:
            self.send_header("Content-Range", f"bytes {start}-{end}/{size}")
        self.end_headers()

        if not send_body:
            return status, 0

        sent = 0
        with file_path.open("rb") as fh:
            fh.seek(start)
            remaining = content_len
            while remaining > 0:
                chunk = fh.read(min(1024 * 1024, remaining))
                if not chunk:
                    break
                self.wfile.write(chunk)
                sent += len(chunk)
                remaining -= len(chunk)
        return status, sent

    def _add_cors_headers(self) -> None:
        cors = self.server.cors_origin  # type: ignore[attr-defined]
        if cors:
            self.send_header("Access-Control-Allow-Origin", cors)
            self.send_header("Access-Control-Allow-Methods", "GET,HEAD,OPTIONS")
            self.send_header("Access-Control-Allow-Headers", "*")

    def _write_access_log(self, status: int, size: int, elapsed_ms: float) -> None:
        remote = self.client_address[0] if self.client_address else "-"
        if remote.startswith("::ffff:"):
            remote = remote[7:]
        ts = clf_timestamp(dt.datetime.now().astimezone())
        line = (
            f'{remote} - - [{ts}] "{self.command} {self.path} HTTP/{self.request_version.split("/", 1)[-1]}" '
            f"{status} {size} {elapsed_ms:.2f}ms"
        )
        print(line)
        log_fp: Optional[TextIO] = self.server.log_fp  # type: ignore[attr-defined]
        if log_fp:
            log_fp.write(line + "\n")
            log_fp.flush()


def main() -> int:
    args = parse_args()
    if args.directory:
        root = Path(args.directory).expanduser().resolve()
    else:
        root = Path.cwd() / "sample_static_site"
        ensure_sample_directory(root)

    if not root.exists() or not root.is_dir():
        print(f"Error: directory does not exist or is not a directory: {root}", file=sys.stderr)
        return 1

    log_fp: Optional[TextIO] = None
    if args.log_path:
        log_fp = open(args.log_path, "a", encoding="utf-8")

    class Server(ThreadingHTTPServer):
        root_real = root.resolve()
        no_listing = args.no_listing
        cache_max_age = args.cache
        cors_origin = args.cors
        log_fp = log_fp

    try:
        httpd = Server((args.host, args.port), StaticHandler)
    except OSError as exc:
        if exc.errno in (98, 10048):
            print(f"Error: port {args.port} is already in use.", file=sys.stderr)
        elif exc.errno in (13, 10013):
            print(f"Error: permission denied binding {args.host}:{args.port}.", file=sys.stderr)
        else:
            print(f"Error: {exc}", file=sys.stderr)
        if log_fp:
            log_fp.close()
        return 1

    if args.ssl:
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain(certfile=args.cert, keyfile=args.key)
        httpd.socket = context.wrap_socket(httpd.socket, server_side=True)

    proto = "https" if args.ssl else "http"
    print("Static HTTP File Server started")
    print(f"URL: {proto}://{args.host}:{args.port}")
    print(f"Serving directory: {root}")
    print(f"Directory listing: {'disabled' if args.no_listing else 'enabled'}")
    print(f"Caching max-age: {args.cache}")
    print(f"CORS: {'enabled (' + args.cors + ')' if args.cors else 'disabled'}")
    print(f"TLS: {'enabled' if args.ssl else 'disabled'}")
    if not args.directory:
        print(
            f"Sample URLs: {proto}://{args.host}:{args.port}/ "
            f"{proto}://{args.host}:{args.port}/assets/style.css "
            f"{proto}://{args.host}:{args.port}/scripts/app.js "
            f"{proto}://{args.host}:{args.port}/sample.svg "
            f"{proto}://{args.host}:{args.port}/hello.txt"
        )

    should_stop = False

    def handle_sigint(signum, frame):  # type: ignore[no-untyped-def]
        nonlocal should_stop
        if not should_stop:
            should_stop = True
            print("\nReceived SIGINT, shutting down gracefully...")
            httpd.shutdown()

    signal.signal(signal.SIGINT, handle_sigint)

    try:
        httpd.serve_forever()
    finally:
        httpd.server_close()
        if log_fp:
            log_fp.close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())