Skip to content

server

server

Standalone HTTP gate server wrapping git http-backend with token auth.

This module has zero terok imports. It is a self-contained security component equivalent to git daemon: a separate process that serves repos.

Token validation

Each request must carry HTTP Basic Auth with the token as the username (password is ignored). The token is looked up in a JSON file mapping tokens to project IDs. The requested repo must match the token's project.

Modes

--inetd Handle one request on an inherited socket (fd 0), then exit. Used by systemd Accept=yes socket activation. --detach Bind, fork, accept loop in child. Daemon fallback.

TokenStore(token_file, *, admin_token=None)

Read-only view of tokens.json with lazy reload on mtime change.

Initialize with the path to the tokens JSON file.

When admin_token is set, it grants access to all repos, bypassing the per-task project scope. Intended for host-level access in containerised deployments.

Source code in src/terok_sandbox/gate/server.py
def __init__(self, token_file: Path, *, admin_token: str | None = None) -> None:
    """Initialize with the path to the tokens JSON file.

    When *admin_token* is set, it grants access to **all** repos,
    bypassing the per-task project scope.  Intended for host-level
    access in containerised deployments.
    """
    self._path = token_file
    self._mtime_ns: int = 0
    self._tokens: dict[str, dict[str, str]] = {}
    self._admin_token = admin_token

validate(token)

Return project_id if token is valid, else None.

Returns :data:_ADMIN_WILDCARD ("*") for admin tokens, granting access to any repo. Reloads the token file when its mtime changes.

Source code in src/terok_sandbox/gate/server.py
def validate(self, token: str) -> str | None:
    """Return project_id if *token* is valid, else ``None``.

    Returns :data:`_ADMIN_WILDCARD` (``"*"``) for admin tokens,
    granting access to any repo.  Reloads the token file when its
    mtime changes.
    """
    if self._admin_token and token == self._admin_token:
        return _ADMIN_WILDCARD
    self._maybe_reload()
    info = self._tokens.get(token)
    if info is None:
        return None
    project = info.get("project")
    return project if isinstance(project, str) else None

main()

Parse CLI args and run the gate server in the selected mode.

Source code in src/terok_sandbox/gate/server.py
def main() -> None:
    """Parse CLI args and run the gate server in the selected mode."""
    parser = argparse.ArgumentParser(
        prog="terok-gate",
        description="HTTP gate server for git repos with token authentication",
    )
    parser.add_argument("--base-path", required=True, help="Root directory for git repos")
    parser.add_argument("--token-file", required=True, help="Path to tokens.json")
    parser.add_argument("--port", type=int, default=9418, help="Listen port (daemon mode)")
    parser.add_argument(
        "--bind",
        default=_LISTEN_ADDR,
        help=f"Address to bind to (default: {_LISTEN_ADDR}). "
        "Use 0.0.0.0 to accept connections from outside.",
    )
    parser.add_argument("--inetd", action="store_true", help="Handle one request on fd 0, exit")
    parser.add_argument("--detach", action="store_true", help="Fork and run as daemon")
    parser.add_argument("--pid-file", default=None, help="PID file path (daemon mode)")
    parser.add_argument(
        "--admin-token",
        default=None,
        help="Admin token granting access to all repos. "
        "Falls back to TEROK_GATE_ADMIN_TOKEN env var.",
    )

    args = parser.parse_args()
    if args.inetd and args.detach:
        parser.error("--inetd and --detach are mutually exclusive")
    _configure_logging(daemon=args.detach)

    admin_token = args.admin_token or os.environ.get("TEROK_GATE_ADMIN_TOKEN")
    base_path = Path(args.base_path)
    token_store = TokenStore(Path(args.token_file), admin_token=admin_token)

    if args.inetd:
        _serve_inetd(base_path, token_store)
    elif args.detach:
        pid_file = Path(args.pid_file) if args.pid_file else None
        _serve_daemon(base_path, token_store, args.port, pid_file, bind=args.bind)
    else:
        parser.error("One of --inetd or --detach is required")