Merge pull request #299 from arXiv/ntai/openid-connect-step-1

[WIP] OpenID connect client library - eyeing toward deploying keycloak

.github/workflows/pullreqeust_tests.yaml

@@ -25,10 +25,17 @@ jobs:
       # - name: Check Types
       # TODO The types are in bad shape and need to be fixed
       #   run: poetry run mypy --exclude "test*" -p arxiv
+
+      - name: Install Chrome Driver
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y chromium-browser chromium-chromedriver
+
       - name: Run other tests
         # These tests are split out because their coverage is low
         run: poetry run pytest --cov=arxiv --cov-fail-under=25 arxiv
       #- name: Check Doc Style
       #  run: poetry run pydocstyle --convention=numpy --add-ignore=D401 arxiv
+
       - name: Run App Tests
         run: poetry run python tests/run_app_tests.py

.gitignore

@@ -129,4 +129,5 @@ dist/
 
 fastly_hourly_stats.ini
 
-test.db-journal
+foo.json
+test.db-journal

arxiv/auth/auth_bridge.py

@@ -0,0 +1,33 @@
+from . import domain
+from .legacy.util import _compute_capabilities
+from .user_claims import ArxivUserClaims
+from .legacy.authenticate import instantiate_tapir_user, _get_user_by_user_id
+from ..db import transaction
+from .legacy.sessions import create as legacy_create_session
+from .legacy.cookies import pack as legacy_pack
+
+def populate_user_claims(user_claims: ArxivUserClaims):
+    """
+    Populate the user's claims to the universe
+    """
+    with transaction():
+        passdata = _get_user_by_user_id(user_claims.user_id)
+        d_user, d_auth = instantiate_tapir_user(passdata)
+
+        session: domain.Session = legacy_create_session(d_auth, user=d_user,
+                                                        tracking_cookie=user_claims.session_id)
+        user_claims.update_claims('tapir_session_id', session.session_id)
+
+
+def bake_cookies(user_claims: ArxivUserClaims) -> (str, str):
+
+    cit_cookie = legacy_pack(user_claims.tapir_session_id,
+                              issued_at=user_claims.issued_at,
+                              user_id=user_claims.user_id,
+                              capabilities=_compute_capabilities(
+                                  user_claims.is_admin,
+                                  user_claims.email_verified,
+                                  user_claims.is_god
+                              ))
+
+    return cit_cookie, ArxivUserClaims.to_arxiv_token_string

arxiv/auth/legacy/authenticate.py

@@ -66,6 +66,29 @@ def authenticate(username_or_email: Optional[str] = None,
     except Exception as ex:
         raise AuthenticationFailed() from ex
 
+    return instantiate_tapir_user(passdata)
+
+
+def instantiate_tapir_user(passdata: PassData) -> Tuple[domain.User, domain.Authorizations]:
+    """
+    Make Tapir user data from pass-data
+
+    Parameters
+    ----------
+    passdata : PassData
+
+    Returns
+    -------
+    :class:`domain.User`
+    :class:`domain.Authorizations`
+
+    Raises
+    ------
+    :class:`AuthenticationFailed`
+        Failed to authenticate user with provided credentials.
+    :class:`Unavailable`
+        Unable to connect to DB.
+    """
     db_user, _, db_nick, db_profile = passdata
     user = domain.User(
         user_id=str(db_user.user_id),

arxiv/auth/legacy/util.py

@@ -55,9 +55,13 @@ def drop_all(engine: Engine) -> None:
 
 def compute_capabilities(tapir_user: TapirUser) -> int:
     """Calculate the privilege level code for a user."""
-    return int(sum([2 * tapir_user.flag_edit_users,
-                    4 * tapir_user.flag_email_verified,
-                    8 * tapir_user.flag_edit_system]))
+    return _compute_capabilities(tapir_user.flag_edit_users,
+                                 tapir_user.flag_email_verified,
+                                 tapir_user.flag_edit_system)
+
+def _compute_capabilities(is_admin: int | bool, email_verified: int | bool, is_god: int | bool) -> int:
+    """Calculate the privilege level code for a user."""
+    return int(sum([2 if is_admin else 0, 4 if email_verified else 0, 8 if is_god else 0]))
 
 
 def get_scopes(db_user: TapirUser) -> List[domain.Scope]: