Added full admin access table and endpoint (#300)

* Added full admin access table and endpoint

* Added method for inviting users

* Fixed refreshing kratos cache

* Function for sending emails

* SMTP variables and login

* Removed unused code

* PR comments

* Added SSO provider to the invite users request

* PR comments

* Removed comments

* PR comments

* PR commentz

* PR comments

* chore: pytest

* Submoduled merged

---------

Co-authored-by: andhreljaKern <[email protected]>

Author: lumburovskalina

alembic/versions/eb96f9b82cc1_added_full_admin_table.py

@@ -0,0 +1,54 @@
+"""Added full admin table
+
+Revision ID: eb96f9b82cc1
+Revises: 455cb5890ac1
+Create Date: 2025-04-24 09:12:33.200446
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = "eb96f9b82cc1"
+down_revision = "455cb5890ac1"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "full_admin_access",
+        sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column("email", sa.String(), nullable=True),
+        sa.Column("meta_info", sa.JSON(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("email"),
+        schema="global",
+    )
+
+    op.execute(
+        """
+        INSERT INTO global.full_admin_access (id, email, meta_info)
+        VALUES
+            (gen_random_uuid(), '[email protected]','{}'),
+            (gen_random_uuid(), '[email protected]','{}'),
+            (gen_random_uuid(), '[email protected]','{}'),
+            (gen_random_uuid(), '[email protected]','{}'),
+            (gen_random_uuid(), '[email protected]','{}'),
+            (gen_random_uuid(), '[email protected]','{}'),
+            (gen_random_uuid(), '[email protected]','{}'),
+            (gen_random_uuid(), '[email protected]','{}'),
+            (gen_random_uuid(), '[email protected]','{}')
+        """
+    )
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("full_admin_access", schema="global")
+    # ### end Alembic commands ###

controller/auth/kratos.py

@@ -1,3 +1,5 @@
+from email.mime.text import MIMEText
+import smtplib
 from typing import Union, Any, List, Dict
 import os
 import requests
@@ -13,6 +15,10 @@
 logger.setLevel(logging.DEBUG)
 
 KRATOS_ADMIN_URL = os.getenv("KRATOS_ADMIN_URL")
+SMTP_HOST = os.getenv("SMTP_HOST")
+SMTP_PORT = os.getenv("SMTP_PORT")
+SMTP_USER = os.getenv("SMTP_USER")
+SMTP_PASSWORD = os.getenv("SMTP_PASSWORD")
 
 # user_id -> {"identity" -> full identity, "simple" -> {"id": str, "mail": str, "firstName": str, "lastName": str}}
 # "collected" -> timestamp
@@ -156,7 +162,7 @@ def resolve_user_name_by_id(user_id: str) -> Dict[str, str]:
     i = __get_identity(user_id, False)
     if i:
         i = i["identity"]
-        return i["traits"]["name"]
+        return i["traits"]["name"] if "name" in i["traits"] else None
     return None
 
 
@@ -191,3 +197,69 @@ def resolve_user_name_and_email_by_id(user_id: str) -> dict:
     if i and "traits" in i and i["traits"]:
         return i["traits"]["name"], i["traits"]["email"]
     return None
+
+
+def create_user_kratos(email: str, provider: str = None):
+    payload_registration = {
+        "schema_id": "default",
+        "traits": {"email": email},
+    }
+    if provider:
+        payload_registration["metadata_public"] = {
+            "registration_scope": {
+                "provider_id": provider,
+                "invitation_sso": True,
+            }
+        }
+    response_create = requests.post(
+        f"{KRATOS_ADMIN_URL}/identities",
+        json=payload_registration,
+    )
+    return response_create.json() if response_create.ok else None
+
+
+def delete_user_kratos(user_id: str) -> bool:
+    response_delete = requests.delete(f"{KRATOS_ADMIN_URL}/identities/{user_id}")
+    if response_delete.ok:
+        del KRATOS_IDENTITY_CACHE[user_id]
+        return True
+    return False
+
+
+def get_recovery_link(user_id: str) -> str:
+    payload_recovery_link = {
+        "expires_in": "48h",
+        "identity_id": user_id,
+    }
+    response_link = requests.post(
+        f"{KRATOS_ADMIN_URL}/recovery/link", json=payload_recovery_link
+    )
+    return response_link.json() if response_link.ok else None
+
+
+def email_with_link(to_email: str, recovery_link: str) -> None:
+    msg = MIMEText(
+        f"Welcome! Click the link to complete your account setup:\n\n{recovery_link}"
+    )
+    msg["Subject"] = "You're invited to our app!"
+    msg["From"] = "[email protected]"
+    msg["To"] = to_email
+
+    with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
+        if SMTP_USER and SMTP_PASSWORD:
+            server.ehlo()
+            server.starttls()
+            server.login(SMTP_USER, SMTP_PASSWORD)
+        server.send_message(msg)
+
+
+def check_user_exists(email: str) -> bool:
+    request = requests.get(
+        f"{KRATOS_ADMIN_URL}/identities?preview_credentials_identifier_similar={quote(email)}"
+    )
+    if request.ok:
+        identities = request.json()
+        for i in identities:
+            if i["traits"]["email"].lower() == email.lower():
+                return True
+    return False

controller/auth/manager.py

@@ -1,5 +1,7 @@
-from typing import Any, Dict
+import re
+from typing import Any, Dict, List, Optional
 
+from controller.auth import kratos
 from fastapi import Request
 from exceptions.exceptions import (
     AuthManagerError,
@@ -11,11 +13,14 @@
 from controller.organization import manager as organization_manager
 from submodules.model import enums, exceptions
 from submodules.model.business_objects import organization
+from submodules.model.business_objects.user import check_email_in_full_admin
 from submodules.model.models import Organization, Project, User
 import sqlalchemy
 
 DEV_USER_ID = "741df1c2-a531-43b6-b259-df23bc78e9a2"
 
+EMAIL_RE = re.compile(r"[\w\.-]+@[\w\.-]+\.\w+")
+
 
 def get_organization_id_by_info(info) -> Organization:
     user = get_user_by_info(info)
@@ -152,3 +157,54 @@ def extract_state_info(request: Request, key: str) -> Any:
         return value
 
     return request.state.parsed[key]
+
+
+def check_is_full_admin(request: Any) -> bool:
+    if request.url.hostname == "localhost" and request.url.port == 7051:
+        return True
+    if check_is_admin(request):
+        jwt_decoded: Dict[str, Any] = jwt.decode(
+            request.headers["Authorization"].split(" ")[1],
+            options={"verify_signature": False},
+        )
+        subject: Dict[str, Any] = jwt_decoded["session"]["identity"]
+        if check_email_in_full_admin(subject["traits"]["email"]):
+            return True
+    return False
+
+
+def invite_users(
+    emails: List[str], organization_name: str, provider: Optional[str] = None
+):
+    user_ids = []
+    for email in emails:
+        # Create accounts for the email
+        user = kratos.create_user_kratos(email, provider)
+        if not user:
+            raise AuthManagerError("User creation failed")
+        user_ids.append(user["id"])
+        # Assign the account to the organization
+        user_manager.update_organization_of_user(organization_name, email)
+
+        # Get the recovery link for the email
+        recovery_link = kratos.get_recovery_link(user["id"])
+        if not recovery_link:
+            raise AuthManagerError("Failed to get recovery link")
+
+        # Send the recovery link to the email
+        kratos.email_with_link(email, recovery_link["recovery_link"])
+    return user_ids
+
+
+def check_valid_emails(emails: List[str]):
+    valid_emails = [
+        email
+        for email in emails
+        if is_valid_email(email) and not kratos.check_user_exists(email)
+    ]
+    all_valid = len(valid_emails) == len(emails)
+    return {"valid_emails": valid_emails, "all_valid": all_valid}
+
+
+def is_valid_email(email: str) -> bool:
+    return bool(EMAIL_RE.fullmatch(email))

fast_api/models.py

@@ -501,3 +501,13 @@ class EvaluationRunDeletionBody(BaseModel):
 class SearchQuestionReformulationBody(BaseModel):
     question: StrictStr
     apiKey: StrictStr
+
+
+class InviteUsersBody(BaseModel):
+    emails: List[StrictStr]
+    organization_name: StrictStr
+    provider: Optional[StrictStr] = None
+
+
+class CheckInviteUsersBody(BaseModel):
+    emails: List[StrictStr]

fast_api/routes/misc.py

@@ -1,7 +1,10 @@
+from exceptions.exceptions import AuthManagerError
 from fastapi import APIRouter, Body, Request, status
 from fastapi.responses import PlainTextResponse
 from fast_api.models import (
     CancelTaskBody,
+    CheckInviteUsersBody,
+    InviteUsersBody,
     ModelProviderDeleteModelBody,
     ModelProviderDownloadModelBody,
     CreateCustomerButton,
@@ -271,3 +274,25 @@ def update_customer_buttons(
             update_request.visible,
         )
     )
+
+
+@router.get("/is-full-admin")
+def get_is_full_admin(request: Request) -> Dict:
+    data = auth.check_is_full_admin(request)
+    return pack_json_result(data)
+
+
+@router.post("/invite-users")
+def invite_users(request: Request, body: InviteUsersBody = Body(...)):
+    if not auth.check_is_full_admin(request):
+        raise AuthManagerError("Full admin access required")
+    data = auth.invite_users(body.emails, body.organization_name, body.provider)
+    return pack_json_result(data)
+
+
+@router.post("/check-valid-emails")
+def check_valid_emails(request: Request, body: CheckInviteUsersBody = Body(...)):
+    if not auth.check_is_full_admin(request):
+        raise AuthManagerError("Full admin access required")
+    data = auth.check_valid_emails(body.emails)
+    return pack_json_result(data)

fast_api/routes/organization.py

@@ -1,4 +1,5 @@
 import json
+from controller.auth import kratos
 from fastapi import APIRouter, Request, Body
 from fast_api.models import (
     AddUserToOrganizationBody,
@@ -90,6 +91,7 @@ def get_user_info(request: Request):
 # in use cognition-ui & admin dashboard (07.01.25)
 @router.get("/get-user-info-extended")
 def get_user_info_extended(request: Request):
+    kratos.__refresh_identity_cache()
     user = auth_manager.get_user_by_info(request.state.info)
     name = resolve_user_name_by_id(user.id)
     user_dict = {
@@ -98,8 +100,8 @@ def get_user_info_extended(request: Request):
             column_whitelist=USER_INFO_WHITELIST,
             column_rename_map=USER_INFO_RENAME_MAP,
         ),
-        "first_name": name.get("first"),
-        "last_name": name.get("last"),
+        "first_name": name.get("first") if name else None,
+        "last_name": name.get("last") if name else None,
     }
 
     return pack_json_result(user_dict)

start

@@ -117,6 +117,8 @@ docker run -d --rm \
 -e SECRET_KEY=default \
 -e POSTGRES_POOL_USE_LIFO=x \
 -e KERN_S3_ENDPOINT=${MINIO_ENDPOINT} \
+-e SMTP_HOST=mailhog \
+-e SMTP_PORT=1025 \
 -v "$INFERENCE_DIR":/inference \
 -v "$LOG_DIR":/logs \
 -v "$CONFIG_DIR":/config \

submodules/model

@@ -0,0 +1,75 @@
+from fastapi.testclient import TestClient
+from controller.auth.kratos import delete_user_kratos
+
+from submodules.model.models import Organization
+import requests
+import time
+
+
+def test_is_full_admin(client: TestClient):
+    """
+    Test validate user is an administrator
+
+    Args:
+        client (TestClient): The test client for making API requests.
+    """
+    response = client.get("/api/v1/misc/is-full-admin")
+    assert response.status_code == 200
+    response_data = response.json()
+
+    assert response_data is True
+
+
+def test_valid_emails(client: TestClient):
+    valid_emails_to_test = ["[email protected]", "[email protected]"]
+    response = client.post(
+        "/api/v1/misc/check-valid-emails",
+        json={"emails": valid_emails_to_test},
+    )
+    assert response.status_code == 200
+    response_data = response.json()
+
+    assert response_data["allValid"] is True
+    assert len(response_data["validEmails"]) == len(valid_emails_to_test)
+
+
+def test_invalid_emails(client: TestClient):
+    valid_emails_to_test = ["[email protected]", "[email protected]"]
+    invalid_emails_to_test = ["test.kern.ai", "devtools@kern"]
+    response = client.post(
+        "/api/v1/misc/check-valid-emails",
+        json={"emails": valid_emails_to_test + invalid_emails_to_test},
+    )
+    assert response.status_code == 200
+    response_data = response.json()
+
+    assert response_data["allValid"] is False
+    assert len(response_data["validEmails"]) == len(valid_emails_to_test)
+
+
+def test_invite_users(client: TestClient, org: Organization):
+    requests.delete("http://mailhog:8025/api/v1/messages")
+    valid_emails_to_test = ["[email protected]"]
+    response = client.post(
+        "/api/v1/misc/invite-users",
+        json={"organization_name": org.name, "emails": valid_emails_to_test},
+    )
+    assert response.status_code == 200
+    created_user_ids = response.json()
+
+    email_response_data = {"total": 0}
+    start_time = time.time()
+    while email_response_data["total"] == 0 and time.time() - start_time < 5:
+        email_response = requests.get(
+            "http://mailhog:8025/api/v2/search",
+            params={"kind": "to", "query": "[email protected]"},
+        )
+        email_response_data = email_response.json()
+    assert email_response.status_code == 200
+
+    for user_id in created_user_ids:
+        delete_user_kratos(user_id)
+
+    assert len(email_response_data["items"]) == len(valid_emails_to_test)
+    assert email_response_data["total"] == len(valid_emails_to_test)
+    assert email_response_data["count"] == len(valid_emails_to_test)