Fix API token expiration and org member getOrganization access

packages/python/app/auth.py

@@ -1,6 +1,7 @@
 # Standard library imports
 import os
 import logging
+from datetime import UTC, datetime, timedelta
 from typing import Optional, Tuple
 from contextlib import asynccontextmanager
 from bson import ObjectId
@@ -48,6 +49,26 @@ def get_api_context(path: str) -> tuple[str, Optional[str]]:
         return "organization", parts[3]
     return "unknown", None
 
+def is_access_token_expired(stored_token: dict) -> bool:
+    """Return True when ``lifetime`` (days) has elapsed since ``created_at``. 0 = no expiry."""
+    lifetime = stored_token.get("lifetime") or 0
+    try:
+        lifetime_days = int(lifetime)
+    except (TypeError, ValueError):
+        lifetime_days = 0
+    if lifetime_days <= 0:
+        return False
+
+    created_at = stored_token.get("created_at")
+    if not isinstance(created_at, datetime):
+        return False
+    if created_at.tzinfo is None:
+        created_at = created_at.replace(tzinfo=UTC)
+
+    expires_at = created_at + timedelta(days=lifetime_days)
+    return datetime.now(UTC) >= expires_at
+
+
 def extract_org_id_from_path(path: str) -> Optional[str]:
     """
     Extract organization ID from the path.
@@ -141,6 +162,9 @@ async def get_current_user(
     stored_token = await db.access_tokens.find_one(token_query)
 
     if stored_token:
+        if is_access_token_expired(stored_token):
+            raise HTTPException(status_code=401, detail="API token expired")
+
         # Validate that user_id from stored token exists in database
         user = await db.users.find_one({"_id": ObjectId(stored_token["user_id"])})
         if not user:
@@ -291,5 +315,7 @@ async def get_org_id_from_token(token: str) -> Optional[str]:
     )
     if not stored_token:
         raise HTTPException(status_code=401, detail="Invalid token")
-
+    if is_access_token_expired(stored_token):
+        raise HTTPException(status_code=401, detail="API token expired")
+
     return stored_token.get("organization_id")

packages/python/app/routes/orgs.py

@@ -24,6 +24,7 @@
     get_current_user,
     is_system_admin,
     is_organization_admin,
+    is_organization_member,
 )
 from app.models import User
 from app.routes.payments import sync_customer, delete_payments_customer
@@ -150,14 +151,14 @@ async def list_organizations(
         if not organization:
             raise HTTPException(status_code=404, detail="Organization not found")
 
-        # Check permissions
-        is_org_admin = await is_organization_admin(organization_id, current_user.user_id)
-
-        if not (is_sys_admin or is_org_admin):
-            raise HTTPException(
-                status_code=403,
-                detail="Not authorized to view this organization"
-            )
+        # Check permissions: system admins, org admins, and org members may fetch by id
+        if not is_sys_admin:
+            is_org_member = await is_organization_member(organization_id, current_user.user_id)
+            if not is_org_member:
+                raise HTTPException(
+                    status_code=403,
+                    detail="Not authorized to view this organization"
+                )
 
         ocr_catalog = await _organization_ocr_catalog()
         return ListOrganizationsResponse(organizations=[

packages/python/tests/test_access_tokens.py

@@ -2,6 +2,7 @@
 from bson import ObjectId
 import os
 import logging
+from datetime import datetime, UTC, timedelta
 
 # Import shared test utilities
 from .conftest_utils import (
@@ -388,4 +389,44 @@ async def test_get_organization_from_token(test_db, mock_auth):
     finally:
         pass  # mock_auth fixture handles cleanup
 
-    logger.info(f"test_get_organization_from_token() end") 
+    logger.info(f"test_get_organization_from_token() end")
+
+
+@pytest.mark.asyncio
+async def test_expired_access_token_rejected(test_db, mock_auth):
+    """Expired API tokens must be rejected at authentication time."""
+    token_data = {
+        "name": "Short-lived Token",
+        "lifetime": 30,
+    }
+
+    create_response = client.post(
+        f"/v0/orgs/{TEST_ORG_ID}/access_tokens",
+        json=token_data,
+        headers=get_auth_headers(),
+    )
+    assert create_response.status_code == 200
+    token_result = create_response.json()
+    api_token = token_result["token"]
+    token_id = token_result["id"]
+
+    await test_db.access_tokens.update_one(
+        {"_id": ObjectId(token_id)},
+        {"$set": {"created_at": datetime.now(UTC) - timedelta(days=31)}},
+    )
+
+    from app.main import app
+
+    original_overrides = app.dependency_overrides.copy()
+    app.dependency_overrides.clear()
+
+    try:
+        docs_response = client.get(
+            f"/v0/orgs/{TEST_ORG_ID}/documents",
+            headers=get_token_headers(api_token),
+        )
+        assert docs_response.status_code == 401
+        assert docs_response.json()["detail"] == "API token expired"
+    finally:
+        app.dependency_overrides = original_overrides
+        await test_db.access_tokens.delete_one({"_id": ObjectId(token_id)})

packages/python/tests/test_org_permissions.py

@@ -238,4 +238,25 @@ async def test_enterprise_creation_restriction(test_db, mock_auth):
 
     except Exception as e:
         logger.error(f"test_enterprise_creation_restriction() failed: {e}")
-        raise
+        raise
+
+
+@pytest.mark.asyncio
+async def test_org_member_can_get_organization_by_id(org_and_users, mock_auth):
+    """Non-admin org members can fetch their organization via organization_id query."""
+    org_id = org_and_users["org_id"]
+    member = org_and_users["member"]
+
+    from app.main import app
+
+    app.dependency_overrides.clear()
+
+    response = client.get(
+        f"/v0/account/organizations?organization_id={org_id}",
+        headers=get_token_headers(member["account_token"]),
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data["organizations"]) == 1
+    assert data["organizations"][0]["id"] == org_id