fmt (black): fatcat_web/

1 files changed, 73 insertions, 57 deletions

diff --git a/python/fatcat_web/auth.py b/python/fatcat_web/auth.py
index 137bc2bb..4fe85770 100644
--- a/python/fatcat_web/auth.py
+++ b/python/fatcat_web/auth.py
@@ -1,4 +1,3 @@
-
 from collections import namedtuple
 
 import fatcat_openapi_client
@@ -12,11 +11,12 @@ from fatcat_web import Config, api, app, login_manager, priv_api
 
 def handle_logout():
     logout_user()
-    for k in ('editor', 'api_token'):
+    for k in ("editor", "api_token"):
         if k in session:
             session.pop(k)
     session.clear()
 
+
 def handle_token_login(token):
     try:
         m = pymacaroons.Macaroon.deserialize(token)
@@ -33,22 +33,23 @@ def handle_token_login(token):
     for caveat in m.first_party_caveats():
         caveat = caveat.caveat_id
         if caveat.startswith(b"editor_id = "):
-            editor_id = caveat[12:].decode('utf-8')
+            editor_id = caveat[12:].decode("utf-8")
     if not editor_id:
         app.log.warning("auth fail: editor_id missing in macaroon")
         abort(400)
     # fetch editor info
     editor = api.get_editor(editor_id)
     session.permanent = True  # pylint: disable=assigning-non-slot
-    session['api_token'] = token
-    session['editor'] = editor.to_dict()
+    session["api_token"] = token
+    session["editor"] = editor.to_dict()
     login_user(load_user(editor.editor_id))
     rp = "/auth/account"
-    if session.get('next'):
-        rp = session['next']
-        session.pop('next')
+    if session.get("next"):
+        rp = session["next"]
+        session.pop("next")
     return redirect(rp)
 
+
 # This will need to login/signup via fatcatd API, then set token in session
 def handle_oauth(remote, token, user_info):
     if user_info:
@@ -57,22 +58,24 @@ def handle_oauth(remote, token, user_info):
         # SUB is the stable internal identifier for the user (not usually the username itself)
         # TODO: should have the real sub here
         # TODO: would be nicer to pass preferred_username for account creation
-        iss = remote.OAUTH_CONFIG['api_base_url']
+        iss = remote.OAUTH_CONFIG["api_base_url"]
 
         # we reuse 'preferred_username' for account name auto-creation (but
         # don't store it otherwise in the backend, at least currently). But i'm
         # not sure all loginpass backends will set it
-        if user_info.get('preferred_username'):
-            preferred_username = user_info['preferred_username']
-        elif 'orcid.org' in iss:
+        if user_info.get("preferred_username"):
+            preferred_username = user_info["preferred_username"]
+        elif "orcid.org" in iss:
             # as a special case, prefix ORCiD identifier so it can be used as a
             # username. If we instead used the human name, we could have
             # collisions. Not a great user experience either way.
-            preferred_username = 'i' + user_info['sub'].replace('-', '')
+            preferred_username = "i" + user_info["sub"].replace("-", "")
         else:
-            preferred_username = user_info['sub']
+            preferred_username = user_info["sub"]
 
-        params = fatcat_openapi_client.AuthOidc(remote.name, user_info['sub'], iss, preferred_username)
+        params = fatcat_openapi_client.AuthOidc(
+            remote.name, user_info["sub"], iss, preferred_username
+        )
         # this call requires admin privs
         (resp, http_status, http_headers) = priv_api.auth_oidc_with_http_info(params)
         editor = resp.editor
@@ -80,90 +83,103 @@ def handle_oauth(remote, token, user_info):
 
         # write token and username to session
         session.permanent = True  # pylint: disable=assigning-non-slot
-        session['api_token'] = api_token
-        session['editor'] = editor.to_dict()
+        session["api_token"] = api_token
+        session["editor"] = editor.to_dict()
 
         # call login_user(load_user(editor_id))
         login_user(load_user(editor.editor_id))
         rp = "/auth/account"
-        if session.get('next'):
-            rp = session['next']
-            session.pop('next')
+        if session.get("next"):
+            rp = session["next"]
+            session.pop("next")
         return redirect(rp)
 
     # XXX: what should this actually be?
     raise Exception("didn't receive OAuth user_info")
 
+
 def handle_ia_xauth(email, password):
-    resp = requests.post(Config.IA_XAUTH_URI,
-        params={'op': 'authenticate'},
+    resp = requests.post(
+        Config.IA_XAUTH_URI,
+        params={"op": "authenticate"},
         json={
-            'version': '1',
-            'email': email,
-            'password': password,
-            'access': Config.IA_XAUTH_CLIENT_ID,
-            'secret': Config.IA_XAUTH_CLIENT_SECRET,
-        })
-    if resp.status_code == 401 or (not resp.json().get('success')):
+            "version": "1",
+            "email": email,
+            "password": password,
+            "access": Config.IA_XAUTH_CLIENT_ID,
+            "secret": Config.IA_XAUTH_CLIENT_SECRET,
+        },
+    )
+    if resp.status_code == 401 or (not resp.json().get("success")):
         try:
-            flash("Internet Archive email/password didn't match: {}".format(resp.json()['values']['reason']))
+            flash(
+                "Internet Archive email/password didn't match: {}".format(
+                    resp.json()["values"]["reason"]
+                )
+            )
         except Exception:
             app.log.warning("IA XAuth fail: {}".format(resp.content))
-        return render_template('auth_ia_login.html', email=email), resp.status_code
+        return render_template("auth_ia_login.html", email=email), resp.status_code
     elif resp.status_code != 200:
         flash("Internet Archive login failed (internal error?)")
         app.log.warning("IA XAuth fail: {}".format(resp.content))
-        return render_template('auth_ia_login.html', email=email), resp.status_code
+        return render_template("auth_ia_login.html", email=email), resp.status_code
 
     # Successful login; now fetch info...
-    resp = requests.post(Config.IA_XAUTH_URI,
-        params={'op': 'info'},
+    resp = requests.post(
+        Config.IA_XAUTH_URI,
+        params={"op": "info"},
         json={
-            'version': '1',
-            'email': email,
-            'access': Config.IA_XAUTH_CLIENT_ID,
-            'secret': Config.IA_XAUTH_CLIENT_SECRET,
-        })
+            "version": "1",
+            "email": email,
+            "access": Config.IA_XAUTH_CLIENT_ID,
+            "secret": Config.IA_XAUTH_CLIENT_SECRET,
+        },
+    )
     if resp.status_code != 200:
         flash("Internet Archive login failed (internal error?)")
         app.log.warning("IA XAuth fail: {}".format(resp.content))
-        return render_template('auth_ia_login.html', email=email), resp.status_code
-    ia_info = resp.json()['values']
+        return render_template("auth_ia_login.html", email=email), resp.status_code
+    ia_info = resp.json()["values"]
 
     # and pass off "as if" we did OAuth successfully
-    FakeOAuthRemote = namedtuple('FakeOAuthRemote', ['name', 'OAUTH_CONFIG'])
-    remote = FakeOAuthRemote(name='archive', OAUTH_CONFIG={'api_base_url': Config.IA_XAUTH_URI})
+    FakeOAuthRemote = namedtuple("FakeOAuthRemote", ["name", "OAUTH_CONFIG"])
+    remote = FakeOAuthRemote(name="archive", OAUTH_CONFIG={"api_base_url": Config.IA_XAUTH_URI})
     oauth_info = {
-        'preferred_username': ia_info['itemname'].replace('@', ''),
-        'iss': Config.IA_XAUTH_URI,
-        'sub': ia_info['itemname'],
+        "preferred_username": ia_info["itemname"].replace("@", ""),
+        "iss": Config.IA_XAUTH_URI,
+        "sub": ia_info["itemname"],
     }
     return handle_oauth(remote, None, oauth_info)
 
+
 def handle_wmoauth(username):
     # pass off "as if" we did OAuth successfully
-    FakeOAuthRemote = namedtuple('FakeOAuthRemote', ['name', 'OAUTH_CONFIG'])
-    remote = FakeOAuthRemote(name='wikipedia', OAUTH_CONFIG={'api_base_url': "https://www.mediawiki.org/w"})
-    conservative_username = ''.join(filter(str.isalnum, username))
+    FakeOAuthRemote = namedtuple("FakeOAuthRemote", ["name", "OAUTH_CONFIG"])
+    remote = FakeOAuthRemote(
+        name="wikipedia", OAUTH_CONFIG={"api_base_url": "https://www.mediawiki.org/w"}
+    )
+    conservative_username = "".join(filter(str.isalnum, username))
     oauth_info = {
-        'preferred_username': conservative_username,
-        'iss': "https://www.mediawiki.org/w",
-        'sub': username,
+        "preferred_username": conservative_username,
+        "iss": "https://www.mediawiki.org/w",
+        "sub": username,
     }
     return handle_oauth(remote, None, oauth_info)
 
+
 @login_manager.user_loader
 def load_user(editor_id):
     # looks for extra info in session, and updates the user object with that.
     # If session isn't loaded/valid, should return None
-    if (not session.get('editor')) or (not session.get('api_token')):
+    if (not session.get("editor")) or (not session.get("api_token")):
         return None
-    editor = session['editor']
-    token = session['api_token']
+    editor = session["editor"]
+    token = session["api_token"]
     user = UserMixin()
     user.id = editor_id
     user.editor_id = editor_id
-    user.username = editor['username']
-    user.is_admin = editor['is_admin']
+    user.username = editor["username"]
+    user.is_admin = editor["is_admin"]
     user.token = token
     return user