update for Werkzeug 3.0 compatibility

http_basic_cookie_auth.py

#!/usr/bin/env python3
#
# Provides an HTTP server for nginx's http_auth_request_module that performs basic authentication
# and sets a cookie that offers longer-lived sessions than regular basic authentication.
#
# Requires argon2-cffi and werkzeug.
#
# Example of configuration (default filename: "http_basic_cookie_auth.json"):
#
# {
#   "listen_addr": "127.0.0.1",
#   "listen_port": 8090,
#   "cookie_name": "permissionbiscuit",
#   "cookie_value": "UQ6nT12DdZVNex2tdG5nxKvMIg5cfnHJQIYhldcauK83o7AOi44v3sXnaBhC20JN",
#   "cookie_duration_s": 7776000,
#   "auth_realm": "kingdom",
#   "basic_username": "secret",
#   "basic_password_hash": "$argon2id$v=19$m=65536,t=3,p=4$nE6tBRRE/zVQ0/2xyoam0Q$ioHYHNxgoiEMt+v94BIBUROJ0V+Q0zzBOnXbg1P/5H4"
# }
#
# (To spoil the fun, the password in this example is "hunter2".)
#
# To hash the basic password, use the following Python snippet:
#
# import getpass
# import argon2
# ph = argon2.PasswordHasher()
# pw = getpass.getpass()
# print(ph.hash(pw))
#
# Integration into nginx looks something like this:
#
# location /private/ {
#   auth_request @privateauth;
#
#   # pass Set-Cookie headers from authenticator to served page
#   auth_request_set $new_cookie $sent_http_set_cookie;
#   add_header Set-Cookie $new_cookie;
# }
# location = @privateauth {
#   internal;
#   proxy_pass http://127.0.0.1:8090/;
#   proxy_pass_request_body off;
#   proxy_set_header Content-Length "";
# }
#
import argparse
import json
import time
import wsgiref.simple_server
import argon2
from werkzeug import Request, Response
from werkzeug.datastructures import WWWAuthenticate


class AuthResponder:
    def __init__(self, config) -> None:
        self.config = config

    def __call__(self, environ, start_response):
        request = Request(environ)
        response = self.respond(request)
        return response(environ, start_response)

    def get_success(self, set_cookie: bool = False) -> Response:
        resp = Response(("success",), 200, content_type="text/plain")
        if set_cookie:
            resp.set_cookie(
                self.config["cookie_name"],
                self.config["cookie_value"],
                expires=time.time() + self.config["cookie_duration_s"],
            )
        return resp

    def respond(self, request: Request) -> Response:
        # verify cookie
        value = request.cookies.get(self.config["cookie_name"], None)
        if value == self.config["cookie_value"]:
            return self.get_success()

        # verify Basic authorization
        authz = request.authorization
        if authz is not None and authz.type == "basic" and authz.username == self.config["basic_username"]:
            hashy = argon2.PasswordHasher()
            success = True
            try:
                hashy.verify(self.config["basic_password_hash"], authz.password)
            except argon2.exceptions.VerifyMismatchError:
                success = False
            if success:
                return self.get_success(set_cookie=True)

        # failure
        resp = Response(("need password",), 401, content_type="text/plain")
        resp.www_authenticate = WWWAuthenticate(
            "basic",
            {
                "realm": self.config["auth_realm"],
                "charset": "UTF-8",
            },
        )
        return resp


def main():
    parser = argparse.ArgumentParser(
        description=
            "Provides an HTTP server for nginx's http_auth_request_module with some well-placed"
            " cookie magic.",
    )
    parser.add_argument(
        dest="config", type=argparse.FileType("r", encoding="utf-8"), metavar="CONFIG.JSON",
        nargs="?", help="The configuration file.", default=None,
    )
    args = parser.parse_args()
    if args.config is None:
        args.config = open("http_basic_cookie_auth.json", "r", encoding="utf-8")

    with args.config:
        config = json.load(args.config)

    responder = AuthResponder(config)
    server = wsgiref.simple_server.make_server(
        config["listen_addr"],
        config["listen_port"],
        responder,
    )
    server.serve_forever()


if __name__ == "__main__":
    main()