Add /admin endpoint

This patch adds an initial `/admin` endpoint that can be accessed with
HTTP Basic authentication. The endpoint is enabled by setting the
environment variable `JULIA_PKG_SERVER_ADMIN_TOKEN_SHA256` to the SHA256
hash of a token that is used as password for the `admin` user.

This patch adds two initial API endpoints:
 - 'GET /admin': Fetch help message about available admin tasks.
 - 'GET /admin/logs': Stream log messages until closing the connection.
   Debug level messages can be enabled by the query parameter
   `level=debug`.

Manifest.toml

@@ -2,7 +2,7 @@
 
 julia_version = "1.9.4"
 manifest_format = "2.0"
-project_hash = "0e17921cfa9ed921cc24638d3f3b86150bc9af48"
+project_hash = "debb167f2e73e4f0b3acdbb9c6435c1f0ce7a775"
 
 [[deps.ArgTools]]
 uuid = "0dad84c5-d112-42e6-8d28-ef12dabb789f"
@@ -119,9 +119,9 @@ uuid = "d6f4376e-aef5-505a-96c1-9c027394607a"
 
 [[deps.MbedTLS]]
 deps = ["Dates", "MbedTLS_jll", "MozillaCACerts_jll", "NetworkOptions", "Random", "Sockets"]
-git-tree-sha1 = "f512dc13e64e96f703fd92ce617755ee6b5adf0f"
+git-tree-sha1 = "c067a280ddc25f196b5e7df3877c6b226d390aaf"
 uuid = "739be429-bea8-5141-9913-cc70e7f3736d"
-version = "1.1.8"
+version = "1.1.9"
 
 [[deps.MbedTLS_jll]]
 deps = ["Artifacts", "Libdl"]

Project.toml

@@ -4,6 +4,7 @@ authors = ["Stefan Karpinski <[email protected]>", "Elliot Saba <staticfloat@
 version = "0.2.0"
 
 [deps]
+Base64 = "2a0f44e3-6c83-55bd-87e4-b1978d98bd5f"
 Dates = "ade2ca70-3891-5945-98fb-dc099432e06a"
 FilesystemDatastructures = "89f0c457-83d8-4998-bfff-8b4a338c9833"
 Gzip_jll = "be1be57a-8558-53c3-a7e5-50095f79957e"
@@ -21,6 +22,7 @@ Sockets = "6462fe0b-24de-5631-8697-dd941f90decc"
 StructTypes = "856f2bd8-1eba-4b0a-8007-ebc267875bd4"
 TOML = "fa267f1f-6049-4f14-aa54-33bafae1ed76"
 Tar = "a4e569a6-e804-4fa4-b0f3-eef7a1d5b13e"
+URIs = "5c2747f8-b7ea-4ff2-ba2e-563bfd36b1d4"
 
 [compat]
 HTTP = "1"

deployment/.env.example

@@ -18,3 +18,10 @@ DISABLE_TLS=1
 
 # Disable internal DNS names (e.g. the `pkgserver-${region}.cflo.at` ones)
 DISABLE_INTERNAL_DNS=1
+
+# *************************************************************
+# * This must only be enabled when the server is behind HTTPS *
+# *************************************************************
+# Configuring an admin token enables some requests to the /admin endpoint. See
+# the beginning of src/admin.jl for more details.
+# ADMIN_TOKEN_SHA256=<...>

deployment/docker-compose.yml

@@ -19,6 +19,7 @@ services:
             JULIA_PKG_SERVER_STORAGE_SERVERS: "${STORAGE_SERVERS:-https://us-east.storage.juliahub.com,https://kr.storage.juliahub.com}"
             JULIA_PKG_SERVER_FLAVORLESS: "${PKG_SERVER_FLAVORLESS:-false}"
             JULIA_PKG_SERVER_REGISTRY_UPDATE_PERIOD: "${UPDATE_PERIOD:-1}"
+            JULIA_PKG_SERVER_ADMIN_TOKEN_SHA256: "${ADMIN_TOKEN_SHA256:-}"
         labels:
             com.centurylinklabs.watchtower.scope: "pkgserver.jl"
             autoheal: "true"

src/PkgServer.jl

@@ -18,7 +18,9 @@ using Gzip_jll
 include("task_utils.jl")
 include("resource.jl")
 include("meta.jl")
+include("admin.jl")
 include("dynamic.jl")
+
 mutable struct RegistryMeta
     # Upstream registry URL (e.g. "https://github.com/JuliaRegistries/General")
     upstream_url::String
@@ -49,6 +51,7 @@ struct ServerConfig
     storage_servers::Vector{String}
     dotflavors::Vector{String}
     registry_update_period::Float64
+    admin_token_sha256::Union{String, Nothing}
 
     # Default server config constructor
     function ServerConfig(; listen_addr = InetAddr(ip"127.0.0.1", 8000),
@@ -73,6 +76,20 @@ struct ServerConfig
         mkpath(joinpath(storage_root, "temp"))
         # Files get stored into `cache`
         mkpath(joinpath(storage_root, "cache"))
+        # /admin interface requires an admin token which must be configured when starting
+        admin_token_sha256 = get(ENV, "JULIA_PKG_SERVER_ADMIN_TOKEN_SHA256", "")
+        if isempty(admin_token_sha256)
+            admin_token_sha256 = nothing
+        else
+            # Verify that this is a hex formatted sha256 hash sum
+            admin_token_sha256 = lowercase(admin_token_sha256)
+            if !occursin(r"^[a-f0-9]{64}$", admin_token_sha256)
+                @warn "The environment variable JULIA_PKG_SERVER_ADMIN_TOKEN_SHA256 is " *
+                      "configured but isn't a hex formatted SHA256-sum. " *
+                      "Disabling the /admin endpoint."
+                admin_token_sha256 = nothing
+            end
+        end
         return new(
             storage_root,
             listen_addr,
@@ -86,6 +103,7 @@ struct ServerConfig
             sort!(storage_servers),
             dotflavors,
             registry_update_period,
+            admin_token_sha256,
         )
     end
 end
@@ -196,7 +214,11 @@ function start(;kwargs...)
                 serve_robots_txt(http)
                 return
             end
-
+            if startswith(resource, "/admin")
+                handle_admin(http)
+                return
+            end
+
             if resource == "/registries" && !flavorless_mode
                 # If they're asking for just "/registries", inspect headers to figure
                 # out which registry flavor they actually want, and if none is given,

src/admin.jl

@@ -0,0 +1,117 @@
+using Base64: Base64
+using Logging: Logging
+using URIs: URIs
+
+include("admin_logging.jl")
+
+# ************************************************************************************
+# *                                IMPORTANT                                         *
+# * The /admin endpoint must only be configured if the server is secured with HTTPS. *
+# * PkgServer itself only uses HTTP, so this has to be configured externally with    *
+# * e.g. a TLS terminating reverse proxy.                                            *
+# *                                IMPORTANT                                         *
+# ************************************************************************************
+#
+# The `/admin` endpoint is enabled by configuring an admin token. The admin token is used as
+# the password for the `admin` user with HTTP basic authentication. The token can be any
+# string, but it is recommended to generate a long random one. PkgServer only need the
+# SHA256 hash of the token. To tell PkgServer about the token use the environment variable
+# `JULIA_PKG_SERVER_ADMIN_TOKEN_SHA256`.
+#
+# The token and the hash can, for example, be setup as with the following commands:
+# ```
+# openssl rand 32 \
+#     | openssl enc -A -base64 \
+#     | tee admin_token.secret \
+#     | sha256sum - \
+#     | cut -d ' ' -f 1 \
+#     | echo "JULIA_PKG_SERVER_ADMIN_TOKEN_SHA256=$(cat -)" >> .env
+# ```
+# This i) generates a 32-byte admin token, ii) encodes it as base64, iii) prints the token
+# to the file `admin_token.secret`, iv) hashes the token, and v) outputs the resulting
+# environment variable to a `.env` file.
+
+function simple_http_response(http::HTTP.Stream, s::Int, msg::Union{String,Nothing}=nothing)
+    HTTP.setstatus(http, s)
+    if msg === nothing
+        HTTP.setheader(http, "Content-Length" => "0")
+        HTTP.startwrite(http)
+    else
+        HTTP.setheader(http, "Content-Type" => "text/plain")
+        HTTP.setheader(http, "Content-Length" => string(sizeof(msg)))
+        HTTP.startwrite(http)
+        write(http, msg)
+    end
+    return nothing
+end
+
+function invalid_auth(http)
+    msg = "Invalid Authorization header, invalid user, or invalid password.\n"
+    HTTP.setheader(http, "WWW-Authenticate" => "Basic")
+    return simple_http_response(http, 401, msg)
+end
+
+# Matches the base64 encoded data in "Basic <base64 data>"
+const basic_auth_regex = r"(?<=^Basic )(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=|[A-Za-z0-9+/]{4})$"
+const userinfo_regex = r"^(?<username>.+?):(?<token>.+)$"
+
+function handle_admin(http::HTTP.Stream)
+    # If no /admin is disabled tell the user
+    if config.admin_token_sha256 === nothing
+        msg = "The /admin endpoint is disabled for this server " *
+              "since no admin token is configured."
+        return simple_http_response(http, 404, msg)
+    end
+    # Check the Authorization header and extract username and token
+    auth = HTTP.header(http, "Authorization", "")
+    m = match(basic_auth_regex, auth)
+    m === nothing && return invalid_auth(http)
+    userinfo = String(Base64.base64decode(m.match::AbstractString))
+    m = match(userinfo_regex, userinfo)
+    m === nothing && return invalid_auth(http)
+    # We have a username and token; verify
+    if !(m["username"]::AbstractString == "admin" &&
+         bytes2hex(SHA.sha256(m["token"]::AbstractString)) == config.admin_token_sha256::String)
+        return invalid_auth(http)
+    end
+    # All good, open sesame!
+    return handle_admin_authenticated(http)
+end
+
+# If we end up here then the request is authenticated
+function handle_admin_authenticated(http::HTTP.Stream)
+    method = http.message.method
+    uri = URIs.URI(http.message.target)
+    if method == "GET" && uri.path == "/admin"
+        admin_help = """
+        Welcome to the /admin endpoint of the package server.
+        The following admin tasks are currently implemented:
+
+        # `GET /admin`
+
+        Fetch this message.
+
+        # `GET /admin/logs`
+
+        Stream log messages until the connection is closed. Debug level log messages are
+        enabled with the query parameter `level=debug` (default is `level=info`). Color can
+        be disabled with the query parameter `color=false` (default is `color=true`).
+        """
+        return simple_http_response(http, 200, admin_help)
+    elseif method == "GET" && uri.path == "/admin/logs"
+        # uri = URIs.URI(http.message.target)
+        params = URIs.queryparams(uri)
+        color = get(params, "color", "true") == "true"
+        req_level = get(params, "level", "info") == "debug" ? Logging.Debug : Logging.Info
+        HTTP.setheader(http, "Content-Type" => "text/plain")
+        HTTP.startwrite(http)
+        attach(ADMIN_LOGGER) do level, message
+            level < req_level && return
+            write(http, color ? message : remove_colors(message))
+        end
+    else
+        msg = "Unknown /admin endpoint '$(method) $(uri.path)'\n"
+        simple_http_response(http, 404, msg)
+    end
+    return
+end

src/admin_logging.jl

@@ -0,0 +1,115 @@
+using Logging: Logging
+using LoggingExtras: LoggingExtras
+
+# TODO: Perhaps all of this can be upstreamed to LoggingExtras as an AttachableLogger.
+
+mutable struct AdminLogger <: Logging.AbstractLogger
+    const lock::ReentrantLock
+    const active_clients::Base.IdSet{Channel{Tuple{Logging.LogLevel, String}}}
+    @atomic n_clients::Int
+    function AdminLogger()
+        return new(ReentrantLock(), Base.IdSet{Channel{Tuple{Logging.LogLevel, String}}}(), 0)
+    end
+end
+
+const ADMIN_LOGGER = AdminLogger()
+
+# Level filtering is done later in the pipeline
+Logging.min_enabled_level(::AdminLogger) = Logging.BelowMinLevel
+
+Logging.catch_exceptions(::AdminLogger) = true
+
+function root_module(m::Module)
+    while m !== parentmodule(m)
+        m = parentmodule(m)
+    end
+    return m
+end
+
+# Only log if we have clients and if the source module is not HTTP
+function Logging.shouldlog(admin_logger::AdminLogger, _, mod, _...)
+    return @atomic(admin_logger.n_clients) > 0 && root_module(mod) != HTTP
+end
+
+# For every log message we stringify once (using ConsoleLogger's format for
+# now) and then send to all the active clients
+function Logging.handle_message(admin_logger::AdminLogger, level, args...; kwargs...)
+    @atomic(admin_logger.n_clients) == 0 && return
+    # Stringify the message using the ConsoleLogger's format
+    io = IOBuffer()
+    ioc = IOContext(io, :color => true)
+    tmp_logger = Logging.ConsoleLogger(ioc, Logging.BelowMinLevel)
+    Logging.handle_message(tmp_logger, level, args...; kwargs...)
+    msg = String(take!(io))
+    # The work for each client is just a put! on an infinitely sized channel which
+    # should i) be quick (holding the lock is fine) and ii) be non-blocking
+    @lock admin_logger.lock begin
+        for client in admin_logger.active_clients
+            try
+                put!(client, (level, msg))
+            catch
+                # The only failure mode above should be if the channel is closed already
+                @assert !isopen(client)
+                # close(client)
+            end
+        end
+    end
+    return nothing
+end
+
+# Used to store the previous global logging configuration when installing the
+# admin logger so that it can be restored later
+PREVIOUS_GLOBAL_LOGGER::Union{Nothing, Logging.AbstractLogger} = nothing
+
+function attach(f::Function, admin_logger::AdminLogger)
+    # Create the client channel with the callback function `f`
+    taskref = Ref{Task}()
+    client = Channel{Tuple{Logging.LogLevel, String}}(Inf; taskref=taskref) do c
+        for (level, msg) in c
+            try
+                f(level, msg)
+            catch _
+                # TODO: Only swallow write errors?
+                close(c)
+            end
+        end
+    end
+    # Take the lock and attach the client to the logger
+    @lock admin_logger.lock begin
+        # Make sure the logger is installed in the global logger. We could leave it
+        # installed at all times, but since we expect to have clients attached quite rarely
+        # we do this little dance to not cause any overhead during normal operation.
+        if PREVIOUS_GLOBAL_LOGGER === nothing
+            global PREVIOUS_GLOBAL_LOGGER = Logging.global_logger(
+                LoggingExtras.TeeLogger(admin_logger, Logging.global_logger()),
+            )
+        end
+        # Add the client
+        push!(admin_logger.active_clients, client)
+        @atomic admin_logger.n_clients = length(admin_logger.active_clients)
+    end
+    # Block until the client is finished.
+    wait(taskref[])
+    @assert !isopen(client)
+    # Client has disconnected
+    @lock admin_logger.lock begin
+        # Detach this (and any other closed) clients from the logger
+        filter!(isopen, admin_logger.active_clients)
+        @atomic admin_logger.n_clients = length(admin_logger.active_clients)
+        # If there are no active clients the admin logger can be uninstalled
+        if @atomic(admin_logger.n_clients) == 0
+            @assert PREVIOUS_GLOBAL_LOGGER !== nothing
+            Logging.global_logger(PREVIOUS_GLOBAL_LOGGER)
+            global PREVIOUS_GLOBAL_LOGGER = nothing
+        end
+    end
+    return
+end
+
+function remove_colors(str::String)
+    io = IOBuffer(; sizehint = sizeof(str))
+    for x in Iterators.filter(x -> x isa Char, Iterators.map(last, Base.ANSIIterator(str)))
+        write(io, x)
+    end
+    return String(take!(io))
+end

test/runtests.jl

@@ -1,4 +1,4 @@
-using PkgServer, Pkg, TOML, HTTP, JSON3, Tar, Test
+using PkgServer, Pkg, TOML, HTTP, JSON3, Tar, Test, Base64, SHA
 
 # You can either perform the following setup:
 #  - Already-running PkgServer, located at $JULIA_PKG_SERVER