Feature/whoami test page

src/nsls2api/api/v1/user_api.py

@@ -1,16 +1,18 @@
 from typing import Annotated
 
 import fastapi
-from fastapi import Depends
+from fastapi import Depends, Request, HTTPException
 
 from nsls2api.api.models.person_model import DataSessionAccess, Person
 from nsls2api.infrastructure.security import (
     get_current_user,
+    get_settings
 )
 from nsls2api.services import (
     bnlpeople_service,
     person_service,
 )
+from nsls2api.services.ldap_service import get_user_info, shape_ldap_response
 
 router = fastapi.APIRouter()
 
@@ -74,9 +76,24 @@ async def get_person_by_department(department_code: str = "PS"):
 
 
 # TODO: Add back into schema if we decide to use this endpoint.
-@router.get("/person/me", response_model=str, include_in_schema=False)
-async def get_myself(current_user: Annotated[Person, Depends(get_current_user)]):
-    return current_user
+@router.get("/person/me",include_in_schema=True)
+async def get_myself(request: Request):
+    upn = request.headers.get("upn")
+    if not upn:
+        raise HTTPException(status_code=400, detail = "upn not found")
+    settings = get_settings()
+    ldap_info = get_user_info(
+        upn,
+        settings.ldap_server,
+        settings.base_dn,
+        settings.bind_user,
+        settings.ldap_bind_password
+    )
+    if not ldap_info:
+        raise HTTPException(status_code=404, detail="User not found in LDAP")
+
+    shaped_info = shape_ldap_response(ldap_info)
+    return shaped_info
 
 
 @router.get("/data-session/{username}", response_model=DataSessionAccess, tags=["data"])

src/nsls2api/infrastructure/config.py

@@ -2,7 +2,7 @@
 from functools import lru_cache
 from pathlib import Path
 
-from pydantic import HttpUrl, MongoDsn
+from pydantic import HttpUrl, MongoDsn, Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 from nsls2api.infrastructure.logging import logger
@@ -78,6 +78,12 @@ class Settings(BaseSettings):
         extra="ignore",
     )
 
+    #Whoami LDAP settings
+    ldap_server: str = Field(default="ldap://ldapproxy.nsls2.bnl.gov", alias="LDAP_SERVER")
+    base_dn: str = Field(default="dc=bnl,dc=gov", alias="BASE_DN")
+    bind_user: str = Field(default="", alias="BIND_USER")
+    ldap_bind_password: str = Field(default="", alias="LDAP_BIND_PASSWORD")
+
 
 @lru_cache()
 def get_settings() -> Settings:

src/nsls2api/services/ldap_service.py

@@ -0,0 +1,136 @@
+from ldap3 import Server, Connection, ASYNC
+from datetime import datetime, timedelta
+import binascii
+from nsls2api.infrastructure.logging import logger
+
+def to_hex(val):
+
+    if isinstance(val, bytes):
+        return binascii.hexlify(val).decode()
+    return None
+
+def get_user_info(upn, ldap_server, base_dn, bind_user, bind_password):
+    conn = None 
+    try:
+        server = Server(ldap_server)
+        conn = Connection(server, user=bind_user, password=bind_password, auto_bind=True)
+        search_filter = f"(&(objectclass=person)(userPrincipalName={upn}))"
+        conn.search(base_dn, search_filter, attributes=['sAMAccountName'])
+
+        if not conn.entries:
+            logger.warning("No entries found for the given UPN.")
+            return None
+
+        entry = conn.entries[0]
+        username = entry.sAMAccountName.value if 'sAMAccountName' in entry else None
+        if username is None:
+            return None
+
+        search_filter = f"(&(objectclass=posixaccount)(sAMAccountName={username}))"
+        conn.search(base_dn, search_filter, attributes=['*'])
+
+        if not conn.entries:
+            logger.warning("no posix entries found for the given username.")
+            return None
+
+        entry = conn.entries[0]
+        user = dict()
+        for attribute in entry.entry_attributes:
+            value = entry[attribute].value
+            if attribute in ("objectGUID", "objectSid"):
+                user[attribute] = value  # keep as bytes
+            else:
+                user[attribute] = str(value)
+        return user
+    except Exception as e:
+        logger.error(f"LDAP Error: {e}")
+        return None
+    finally:
+        if conn is not None:
+            conn.unbind()
+
+def filetime_to_str(filetime):
+    try:
+        if filetime is None or int(filetime) == 0 or int(filetime) == 9223372036854775807:
+            return "Never"
+        dt = datetime(1601, 1, 1) + timedelta(microseconds=int(filetime) // 10)
+        return dt.strftime("%Y-%m-%d %H:%M:%S UTC")
+    except Exception:
+        return str(filetime)
+
+def generalized_time_to_str(gt):
+    try:
+        if not gt: return ""
+        dt = datetime.strptime(gt.split(".")[0], "%Y%m%d%H%M%S")
+        return dt.strftime("%Y-%m-%d %H:%M:%S UTC")
+    except Exception:
+        return str(gt)
+
+def decode_uac(uac):
+    flags = []
+    try:
+        val = int(uac)
+        if val & 0x0001: flags.append("SCRIPT")
+        if val & 0x0002: flags.append("ACCOUNTDISABLE")
+        if val & 0x0008: flags.append("HOMEDIR_REQUIRED")
+        if val & 0x0200: flags.append("NORMAL_ACCOUNT")
+        if val & 0x1000: flags.append("PASSWORD_EXPIRED")
+    except Exception:
+        return []
+    return flags or ["NORMAL_ACCOUNT"]
+
+def shape_ldap_response(user_info, dn=None, status="Read", read_time=None):
+    def clean_groups(groups_val):
+        if not groups_val:
+            return []
+        if isinstance(groups_val, list):
+            return groups_val
+        elif isinstance(groups_val, str):
+            return [g.strip() for g in groups_val.replace("\n", ",").split(",") if g.strip()]
+        return []
+
+    return {
+        "dn": dn or user_info.get("distinguishedName"),
+        "status": status,
+        "readTime": read_time,
+        "identity": {
+            "displayName": user_info.get("displayName"),
+            "email": user_info.get("mail") or user_info.get("email"),
+            "department": user_info.get("department"),
+            "manager": user_info.get("manager"),
+            "unix": {
+                "uid": user_info.get("uid"),
+                "uidNumber": user_info.get("uidNumber"),
+                "gidNumber": user_info.get("gidNumber"),
+                "homeDirectory": user_info.get("homeDirectory"),
+                "loginShell": user_info.get("loginShell")
+            }
+        },
+        "account": {
+            "accountExpires": filetime_to_str(user_info.get("accountExpires")),
+            "badPasswordTime": filetime_to_str(user_info.get("badPasswordTime")),
+            "badPwdCount": int(user_info.get("badPwdCount") or 0),
+            "pwdLastSet": filetime_to_str(user_info.get("pwdLastSet")),
+            "lastLogon": filetime_to_str(user_info.get("lastLogon")),
+            "userAccountControlFlags": decode_uac(user_info.get("userAccountControl")),
+            "userPrincipalName": user_info.get("userPrincipalName"),
+        },
+        "directory": {
+            "objectGUID": to_hex(user_info.get("objectGUID")),
+            "objectSid": to_hex(user_info.get("objectSid")),
+            "primaryGroupID": user_info.get("primaryGroupID"),
+            "distinguishedName": user_info.get("distinguishedName"),
+        },
+        "groups": clean_groups(user_info.get("memberOf")),
+        "attributes": {
+            "sn": user_info.get("sn"),
+            "givenName": user_info.get("givenName"),
+            "description": user_info.get("description"),
+            "gecos": user_info.get("gecos"),
+            "street": user_info.get("street"),
+            "codePage": user_info.get("codePage"),
+            "countryCode": user_info.get("countryCode"),
+            "instanceType": user_info.get("instanceType"),
+            "objectClass": [s.strip() for s in user_info.get("objectClass", "").split() if s.strip()]
+        }
+    }