Wiki-Login to Basic Auth API

This is a layer that takes a request with basic auth credentials and tries to
log into a MediaWiki using them. It returns a corresponding error code
(401, 403, 200).

This is meant to be used e.g. with nginx auth_request to use a MediaWiki
as the authentication backend.

.gitignore

+__pycache__/
+config.py
+venv/

app.py

+#!/usr/bin/env python3
+
+from flask import Flask, g, request, abort, Response
+
+from wiki import WikiClient, WikiException
+
+import config
+domain = getattr(config, "WIKI_DOMAIN", None)
+
+app = Flask(__name__)
+app.config.from_object(config)
+
+def try_wiki_login(user, password):
+    client = WikiClient(endpoint=config.WIKI_API_ENDPOINT)
+    try:
+        client.login(user, password, domain=domain)
+        client.logout()
+        return True
+    except WikiException as error:
+        print(error)
+    return False
+
+@app.route("/")
+def index():
+    auth = request.authorization
+    if not auth:
+        return Response(
+            "Please authenticate with your Wiki credentials.", 401,
+            {"WWW-Authenticate": "Basic realm=\"Wiki\""})
+    if not try_wiki_login(auth.username, auth.password):
+        return Response("Forbidden", status=403)
+    return Response("OK", status=200)
+            

config.example.py

+
+SECRET_KEY = "<random string>"
+
+WIKI_API_ENDPOINT = "https://wiki.example.com/api.php"
+
+# WIKI_DOMAIN = "domain"

requirements.txt

+certifi==2018.4.16
+chardet==3.0.4
+click==6.7
+Flask==1.0.2
+idna==2.6
+itsdangerous==0.24
+Jinja2==2.10
+MarkupSafe==1.0
+requests==2.18.4
+urllib3==1.22
+Werkzeug==0.14.1

wiki.py

+import requests
+from json import JSONDecodeError
+
+HTTP_STATUS_OK = 200
+HTTP_STATUS_AUTHENTICATE = 401
+
+
+class WikiException(Exception):
+    pass
+
+
+def _filter_params(params):
+    result = {}
+    for key, value in sorted(params.items(), key=lambda t: t[0] == "token"):
+        if isinstance(value, bool):
+            if value:
+                result[key] = ""
+        else:
+            result[key] = value
+    return result
+
+
+class WikiClient:
+    def __init__(self, endpoint):
+        self.endpoint = endpoint
+        self.token = None
+        self.cookies = requests.cookies.RequestsCookieJar()
+
+    def is_logged_in(self):
+        return self.token is not None
+
+    def login(self, user, password, domain=None):
+        # todo: Change this to the new MediaWiki tokens api
+        token_answer = self.do_action("login", method="post", lgname=user)
+        if "login" not in token_answer or "token" not in token_answer["login"]:
+            raise WikiException("No token in login answer.")
+        lgtoken = token_answer["login"]["token"]
+        login_answer = self.do_action(
+            "login", method="post", lgname=user, lgpassword=password,
+            lgdomain=domain, lgtoken=lgtoken)
+        if ("login" not in login_answer
+                or "result" not in login_answer["login"]
+                or login_answer["login"]["result"] != "Success"):
+            raise WikiException("Login not successful.")
+
+    def logout(self):
+        self.do_action("logout")
+
+    def do_action(self, action, method="get", data=None, **kwargs):
+        kwargs["action"] = action
+        kwargs["format"] = "json"
+        params = _filter_params(kwargs)
+
+        def _do_request():
+            if method == "get":
+                return requests.get(
+                    self.endpoint, cookies=self.cookies, params=params)
+            elif method == "post":
+                return requests.post(
+                    self.endpoint, cookies=self.cookies, data=data,
+                    params=params)
+        req = _do_request()
+        if req.status_code != HTTP_STATUS_OK:
+            raise WikiException(
+                "HTTP status code {} on action {}.".format(
+                    req.status_code, action))
+        self.cookies.update(req.cookies)
+        try:
+            return req.json()
+        except JSONDecodeError:
+            raise WikiException("Server did not return valid JSON.")
+