add ResourceType endpoints

ee/api/routers/scim.py

@@ -4,7 +4,7 @@ import uuid
 from typing import Optional
 
 from decouple import config
-from fastapi import Depends, HTTPException, Header, Query, Response
+from fastapi import Depends, HTTPException, Header, Query, Response, Request
 from fastapi.responses import JSONResponse
 from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 from pydantic import BaseModel, Field
@@ -36,7 +36,6 @@ async def login(host: str = Header(..., alias="Host"), form_data: OAuth2Password
     if form_data.username != config("SCIM_USER") or form_data.password != config("SCIM_PASSWORD"):
         raise HTTPException(status_code=401, detail="Invalid credentials")
 
-    subdomain = "Openreplay EE"
     tenant = tenants.get_by_name(subdomain)
     access_token, refresh_token = create_tokens(tenant_id=tenant["tenantId"])
 
@@ -49,7 +48,71 @@ async def refresh_token(r: RefreshRequest):
     payload = verify_refresh_token(r.refresh_token)
     new_access_token, _ = create_tokens(tenant_id=payload["tenant_id"])
 
-    return {"access_token": new_access_token, "token_type": "Bearer"}
+    return {"access_token": new_access_token, "token_type": "bearer"}
+
+
+RESOURCE_TYPE_IDS_TO_RESOURCE_TYPE_DETAILS = {
+    "User": {
+        "schemas": ["urn:ietf:params:scim:schemas:core:2.0:ResourceType"],
+        "id": "User",
+        "name": "User",
+        "endpoint": "/Users",
+        "description": "User account",
+        "schema": "urn:ietf:params:scim:schemas:core:2.0:User",
+    }
+}
+
+
+def _not_found_error_response(resource_id: str):
+    return JSONResponse(
+        status_code=404,
+        content={
+            "schemas": ["urn:ietf:params:scim:api:messages:2.0:Error"],
+            "detail": f"Resource {resource_id} not found",
+            "status": "404",
+        }
+    )
+
+
+@public_app.get("/ResourceTypes", dependencies=[Depends(auth_required)])
+async def get_resource_types(r: Request):
+    return JSONResponse(
+        status_code=200,
+        content={
+            "totalResults": len(RESOURCE_TYPE_IDS_TO_RESOURCE_TYPE_DETAILS),
+            "itemsPerPage": len(RESOURCE_TYPE_IDS_TO_RESOURCE_TYPE_DETAILS),
+            "startIndex": 1,
+            "schemas": ["urn:ietf:params:scim:api:messages:2.0:ListResponse"],
+            "Resources": sorted(
+                {
+                    **value,
+                    "meta": {
+                        "location": str(r.url_for("get_resource_type", resource_id=value["id"])),
+                        "resourceType": "ResourceType",
+                    }
+                }
+                for value in RESOURCE_TYPE_IDS_TO_RESOURCE_TYPE_DETAILS.values()
+            ),
+        },
+    )
+
+
+@public_app.get("/ResourceTypes/{resource_id}", dependencies=[Depends(auth_required)])
+async def get_resource_type(r: Request, resource_id: str):
+    if resource_id not in RESOURCE_TYPE_IDS_TO_RESOURCE_TYPE_DETAILS:
+        return _not_found_error_response(resource_id)
+    content = {
+        **RESOURCE_TYPE_IDS_TO_RESOURCE_TYPE_DETAILS[resource_id],
+        "meta": {
+            "location": str(r.url),
+            "resourceType": "ResourceType",
+        }
+    }
+    return JSONResponse(
+        status_code=200,
+        content=content,
+    )
+
 
 """
 User endpoints