Securing GeoServer REST Credentials in Python Automation

Resolve the credential inside the call rather than holding it as an attribute, read it from a mounted file or the environment rather than from code, refuse Basic authentication over plain HTTP to a remote host, and give automation its own role scoped to what it actually does. Most credential leaks in spatial automation happen through ordinary operation — a traceback, an access log, a committed catalog export — rather than through an attack.

The Core Challenge: The Catalog API Is Root

The GeoServer REST API can create data stores with arbitrary connection parameters, rewrite any style, and delete any layer. A credential for it is not “a password for the map server”; it is administrative access to everything the service publishes and, through store definitions, to the databases behind it.

That raises the cost of the ordinary leak paths. A password in a repository is permanent — rewriting history does not remove it from clones or forks. A password in an instance attribute appears in every traceback that captures the object, which is every traceback in a request-handling path. A password in a URL reaches proxy access logs, server logs and, if the URL is ever emitted in a page, the referrer header. None of these requires anyone to attack anything.

Production-Ready Code

from __future__ import annotations

import os
from dataclasses import dataclass, field
from typing import Any, Callable

import requests
from requests.auth import HTTPBasicAuth

LOCAL_HOSTS = ("http://localhost", "http://127.0.0.1", "http://[::1]")

class CredentialError(RuntimeError):
    pass

@dataclass
class Secret:
    """A credential held as a resolver, never as a value.

    Nothing here stores the plaintext, so it cannot appear in a repr, a
    pickle, a traceback frame or a debugger's locals view. Resolution on
    each call also means a rotated secret takes effect immediately.
    """
    name: str
    _resolve: Callable[[], str | None] = field(repr=False)

    def get(self) -> str:
        value = self._resolve()
        if not value:
            raise CredentialError(f"{self.name} is not set")
        return value

    def __str__(self) -> str:
        return f"<secret {self.name}>"

def from_env(variable: str) -> Secret:
    return Secret(variable, lambda: os.environ.get(variable))

def from_file(path: str) -> Secret:
    def read() -> str | None:
        try:
            with open(path, encoding="utf-8") as fh:
                return fh.read().strip()
        except OSError:
            return None
    return Secret(f"file:{path}", read)

def first_available(*secrets: Secret) -> Secret:
    """Prefer a mounted file, fall back to the environment."""
    def resolve() -> str | None:
        for secret in secrets:
            try:
                return secret.get()
            except CredentialError:
                continue
        return None
    return Secret(" or ".join(s.name for s in secrets), resolve)

@dataclass
class RestClient:
    base: str
    user: str
    secret: Secret
    verify_tls: bool | str = True
    timeout: int = 60

    def __post_init__(self) -> None:
        if self.base.startswith("http://") and not self.base.startswith(LOCAL_HOSTS):
            raise CredentialError(
                f"refusing to send Basic credentials over plain HTTP to "
                f"{self.base!r}")
        if "@" in self.base.split("//", 1)[-1].split("/", 1)[0]:
            raise CredentialError(
                "credentials must not be embedded in the base URL — they "
                "reach every access log on the path")

    def call(self, method: str, path: str, **kwargs: Any) -> requests.Response:
        try:
            resp = requests.request(
                method, f"{self.base.rstrip('/')}{path}",
                auth=HTTPBasicAuth(self.user, self.secret.get()),
                verify=self.verify_tls, timeout=self.timeout, **kwargs)
        except requests.RequestException as exc:
            # Re-raise without the request object, which holds the header.
            raise RuntimeError(f"{method} {path} failed: {type(exc).__name__}") from None
        if resp.status_code in (401, 403):
            raise PermissionError(
                f"{self.user!r} is not permitted to {method} {path}")
        resp.raise_for_status()
        return resp

SECRET_FIELDS = {"passwd", "password", "PASSWORD", "user", "connectionParameters"}

def strip_secrets(node: Any) -> Any:
    """Remove credentials from a catalog payload before it touches disk.

    GeoServer returns store connection parameters, including the password,
    to an administrator. Any export that is committed must be stripped.
    """
    if isinstance(node, dict):
        out: dict[str, Any] = {}
        for key, value in node.items():
            if key in {"passwd", "password", "PASSWORD"}:
                out[key] = "${STORE_PASSWORD}"
            elif key == "connectionParameters" and isinstance(value, dict):
                out[key] = {"entry": [
                    e if e.get("@key") not in {"passwd", "password"}
                    else {"@key": e["@key"], "$": "${STORE_PASSWORD}"}
                    for e in value.get("entry", [])]}
            else:
                out[key] = strip_secrets(value)
        return out
    if isinstance(node, list):
        return [strip_secrets(v) for v in node]
    return node

if __name__ == "__main__":
    client = RestClient(
        base=os.environ["GEOSERVER_URL"],
        user=os.environ.get("GEOSERVER_USER", "automation"),
        secret=first_available(
            from_file("/run/secrets/geoserver_password"),
            from_env("GEOSERVER_PASSWORD")),
    )
    print(client.call("GET", "/rest/about/version.json").json())

Step-by-Step Walkthrough

The secret is a callable, not a string. Secret stores a resolver with repr=False, so the object can be printed, logged, pickled or captured in a traceback frame without revealing anything. This is a small amount of ceremony that removes the most common leak path entirely, and it comes with rotation for free — nothing cached the old value.

Refuse plain HTTP structurally. __post_init__ rejects a remote http:// base outright rather than warning, because a warning in a log is not a control. Localhost is exempted so that development against a container on the same host still works, which is the case that otherwise pushes people to disable the check globally.

Credentials in the URL are rejected too. https://user:pass@host/geoserver is convenient and lands the credential in every proxy access log along the path, plus the server’s own. Detecting the userinfo component and refusing costs one line.

Exceptions are re-raised without the request. A requests exception carries the Request object, which carries the Authorization header. An unhandled traceback therefore prints the encoded credential. Raising a plain RuntimeError with from None drops that chain, at the cost of some diagnostic detail — a trade worth making on a path that runs unattended.

Verification

Check the two properties that matter — that the secret does not appear where it should not, and that the role is scoped:

import pickle, traceback

client = RestClient(base="https://geo.example.org/geoserver",
                    user="automation", secret=from_env("GEOSERVER_PASSWORD"))

assert "GEOSERVER_PASSWORD" in repr(client)
assert os.environ["GEOSERVER_PASSWORD"] not in repr(client)
assert os.environ["GEOSERVER_PASSWORD"] not in str(pickle.dumps(client))

try:
    client.call("DELETE", "/rest/workspaces/production")
except PermissionError as exc:
    print("scoped correctly:", exc)

scoped correctly: 'automation' is not permitted to DELETE /rest/workspaces/production

Then confirm plain HTTP is refused, which is the check most likely to regress when someone adds a staging environment:

import pytest
with pytest.raises(CredentialError, match="plain HTTP"):
    RestClient("http://staging.example.org/geoserver", "automation",
               from_env("GEOSERVER_PASSWORD"))

Gotchas & Edge Cases

Give automation its own role. Using the admin account because it works is how a pipeline that only publishes layers acquires the ability to delete workspaces. A role scoped to the workspaces the automation owns turns a bug in a script into a failed request rather than a data loss event.

Catalog exports carry store passwords. GeoServer returns connection parameters including the password to an administrator, so any tooling that walks the catalog — including the export and restore workflow — must strip them before writing anything. strip_secrets handles both the flat and the connection-parameter forms, which differ between endpoints.

Do not disable TLS verification to make staging work. verify=False silences the error and accepts any certificate, which is the whole protection. Point verify at the internal certificate authority bundle instead — the parameter takes a path, which is why the field is typed as bool | str.

Rotate on suspicion, not on proof. Confirming whether a credential leaked usually takes longer than rotating it. Because resolution happens per call, rotation here is a secret-store update and nothing else — no restart, no deploy, no coordination.

Frequently Asked Questions

---

Back to Security and Access Control for Spatial Services

Related

• Security and Access Control for Spatial Services — the four layers this credential handling sits inside
• Exporting and Restoring a GeoServer Catalog With Python — the export that must be stripped
• Rate Limiting OGC Requests With a Python Reverse Proxy — bounding cost once identity is settled