Metadata client

fixes #6 (closed)

geomagio/api/secure/login.py

@@ -35,6 +35,7 @@ Usage:
 """
 import logging
 import os
+import requests
 from typing import Callable, List, Optional
 
 from authlib.integrations.starlette_client import OAuth
@@ -56,7 +57,8 @@ class User(BaseModel):
 
 
 async def current_user(request: Request) -> Optional[User]:
-    """Get logged in user, or None if not logged in.
+    """Get user information from gitlab access token or session(if currently logged in).
+    Returns none if access token is not vald or user is not logged in.
 
     Usage example:
         user: Optional[User] = Depends(current_user)
@@ -64,9 +66,45 @@ async def current_user(request: Request) -> Optional[User]:
     """
     if "user" in request.session:
         return User(**request.session["user"])
+    if "apiuser" in request.session:
+        return User(**request.session["apiuser"])
+    if "Authorization" in request.headers:
+        user = get_api_user(token=request.headers["Authorization"])
+        if user is not None:
+            request.session["apiuser"] = user.dict()
+            return user
     return None
 
 
+def get_api_user(token: str) -> Optional[User]:
+    url = os.getenv("GITLAB_API_URL")
+    header = {"PRIVATE-TOKEN": token}
+    # request user information from gitlab api with access token
+    userinfo_response = requests.get(
+        f"{url}/user",
+        headers=header,
+    )
+    userinfo = userinfo_response.json()
+    try:
+        user = User(
+            email=userinfo["email"],
+            sub=userinfo["id"],
+            name=userinfo["name"],
+            nickname=userinfo["username"],
+            picture=userinfo["avatar_url"],
+        )
+    except KeyError:
+        logging.info(f"Invalid token: {userinfo_response.status_code}")
+        return None
+    # use valid token to retrieve user's groups
+    groups_response = requests.get(
+        f"{url}/groups",
+        headers=header,
+    )
+    user.groups = [g["full_path"] for g in groups_response.json()]
+    return user
+
+
 def require_user(
     allowed_groups: List[str] = None,
 ) -> Callable[[Request, User], User]: