Release 0.12.0

environment/dev.env

@@ -12,4 +12,6 @@ export HF_MODEL="BAAI/bge-small-en-v1.5"
 
 export GH_CLIENT_ID=`pass databio/pephub/gh_client_id`
 export GH_CLIENT_SECRET=`pass databio/pephub/gh_client_secret`
-export BASE_URI=http://localhost:8000
+export BASE_URI=http://localhost:8000
+
+ export PH_DEV_MODE=true

pephub/_version.py

@@ -1 +1 @@
-__version__ = "0.11.8"
+__version__ = "0.11.9"

pephub/const.py

@@ -123,9 +123,10 @@
     "CRITICAL": logging.CRITICAL,
 }
 
-JWT_SECRET = token_hex(32)
+JWT_SECRET = "0" * 64 if os.getenv("PH_DEV_MODE") is not None else token_hex(32)
 JWT_EXPIRATION = 4320  # 3 days in minutes
 JWT_EXPIRATION_SECONDS = JWT_EXPIRATION * 60  # seconds
+MAX_NEW_KEYS = 5
 
 AUTH_CODE_EXPIRATION = 5 * 60  # seconds
 

pephub/dependencies.py

@@ -1,9 +1,9 @@
 import json
 import logging
 import os
-from datetime import datetime, timedelta
+from datetime import datetime
 from secrets import token_hex
-from typing import Any, Dict, Generator, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 from cachetools import cached, TTLCache
 
 import jwt
@@ -31,10 +31,11 @@
     DEFAULT_POSTGRES_USER,
     DEFAULT_QDRANT_HOST,
     DEFAULT_QDRANT_PORT,
-    JWT_EXPIRATION,
     JWT_SECRET,
 )
+from .helpers import jwt_encode_user_data
 from .routers.models import ForkRequest
+from .developer_keys import dev_key_handler
 
 _LOGGER_PEPHUB = logging.getLogger("uvicorn.access")
 
@@ -83,14 +84,8 @@ def _request_user_data_from_github(access_token: str) -> UserData:
             )
 
     @staticmethod
-    def jwt_encode_user_data(user_data: dict) -> str:
-        exp = datetime.utcnow() + timedelta(minutes=JWT_EXPIRATION)
-        encoded_user_data = jwt.encode(
-            {**user_data, "exp": exp}, JWT_SECRET, algorithm="HS256"
-        )
-        if isinstance(encoded_user_data, bytes):
-            encoded_user_data = encoded_user_data.decode("utf-8")
-        return encoded_user_data
+    def jwt_encode_user_data(user_data: dict, exp: datetime = None) -> str:
+        return jwt_encode_user_data(user_data, exp=exp)
 
 
 # database connection
@@ -132,19 +127,22 @@ def get_db() -> PEPDatabaseAgent:
     return agent
 
 
-def read_authorization_header(Authorization: str = Header(None)) -> Union[dict, None]:
+def read_authorization_header(authorization: str = Header(None)) -> Union[dict, None]:
     """
     Reads and decodes a JWT, returning the decoded variables.
 
     :param Authorization: JWT provided via FastAPI injection from the API cookie.
     """
-    if Authorization is None:
+    if authorization is None:
         return None
     else:
-        Authorization = Authorization.replace("Bearer ", "")
+        authorization = authorization.replace("Bearer ", "")
     try:
         # Python jwt.decode verifies content as well so this is safe.
-        session_info = jwt.decode(Authorization, JWT_SECRET, algorithms=["HS256"])
+        # check last 5 chars
+        if dev_key_handler.is_key_bad(authorization[-5:]):
+            raise HTTPException(401, "JWT has been revoked")
+        session_info = jwt.decode(authorization, JWT_SECRET, algorithms=["HS256"])
     except jwt.exceptions.InvalidSignatureError as e:
         _LOGGER_PEPHUB.error(e)
         return None
@@ -201,7 +199,7 @@ def get_project(
         description="Return the project with the samples pephub_id",
         include_in_schema=False,
     ),
-) -> Dict[str, Any]:
+) -> Dict[str, Any]:  # type: ignore
     try:
         proj = agent.project.get(namespace, project, tag, raw=True, with_id=with_id)
         yield proj
@@ -217,7 +215,7 @@ def get_config(
     project: str,
     tag: Optional[str] = DEFAULT_TAG,
     agent: PEPDatabaseAgent = Depends(get_db),
-) -> Dict[str, Any]:
+) -> Dict[str, Any]:  # type: ignore
     try:
         config = agent.project.get_config(namespace, project, tag)
         yield config
@@ -233,7 +231,7 @@ def get_subsamples(
     project: str,
     tag: Optional[str] = DEFAULT_TAG,
     agent: PEPDatabaseAgent = Depends(get_db),
-) -> Dict[str, Any]:
+) -> Dict[str, Any]:  # type: ignore # type: ignore
     try:
         subsamples = agent.project.get_subsamples(namespace, project, tag)
         yield subsamples
@@ -250,7 +248,7 @@ def get_project_annotation(
     tag: Optional[str] = DEFAULT_TAG,
     agent: PEPDatabaseAgent = Depends(get_db),
     namespace_access_list: List[str] = Depends(get_namespace_access_list),
-) -> AnnotationModel:
+) -> AnnotationModel:  # type: ignore
     try:
         anno = agent.annotation.get(
             namespace, project, tag, admin=namespace_access_list
@@ -320,7 +318,7 @@ def verify_user_can_read_project(
 def verify_user_can_fork(
     fork_request: ForkRequest,
     namespace_access_list: List[str] = Depends(get_namespace_access_list),
-) -> bool:
+) -> bool:  # type: ignore
     fork_namespace = fork_request.fork_to
     if fork_namespace in (namespace_access_list or []):
         yield
@@ -344,7 +342,7 @@ def get_qdrant_enabled() -> bool:
 
 def get_qdrant(
     qdrant_enabled: bool = Depends(get_qdrant_enabled),
-) -> Union[QdrantClient, None]:
+) -> Union[QdrantClient, None]:  # type: ignore
     """
     Return connection to qdrant client
     """
@@ -383,7 +381,7 @@ def get_namespace_info(
     namespace: str,
     agent: PEPDatabaseAgent = Depends(get_db),
     user: str = Depends(get_user_from_session_info),
-) -> Namespace:
+) -> Namespace:  # type: ignore
     """
     Get the information on a namespace, if it exists.
     """

pephub/developer_keys.py

@@ -0,0 +1,81 @@
+from datetime import datetime, timedelta
+from typing import Dict, List
+
+import secrets
+
+from fastapi import HTTPException
+
+from .routers.models import DeveloperKey
+
+from .helpers import jwt_encode_user_data
+from .const import MAX_NEW_KEYS
+
+
+class DeveloperKeyHandler:
+    def __init__(self, default_exp: int = 30 * 24 * 60 * 60):
+        self._keys: Dict[str, List[DeveloperKey]] = {}
+        self._default_exp = default_exp
+        self._bad_jwts = []
+
+    def add_key(self, namespace: str, key: DeveloperKey):
+        """
+        Add a key to the handler for a given namespace/user
+
+        :param namespace: namespace for the key
+        :param key: DeveloperKey object
+        """
+        if namespace not in self._keys:
+            self._keys[namespace] = []
+        if len(self._keys[namespace]) >= MAX_NEW_KEYS:
+            raise HTTPException(
+                status_code=400,
+                detail="You have reached the maximum number of keys allowed",
+            )
+        self._keys[namespace].append(key)
+
+    def get_keys_for_namespace(self, namespace: str) -> List[DeveloperKey]:
+        """
+        Get all the keys for a given namespace
+
+        :param namespace: namespace for the key
+        """
+        return self._keys.get(namespace) or []
+
+    def remove_key(self, namespace: str, last_five_chars: str):
+        """
+        Remove a key from the handler for a given namespace/user
+
+        :param namespace: namespace for the key
+        :param key: key to remove
+        """
+        if namespace in self._keys:
+            self._keys[namespace] = [
+                key for key in self._keys[namespace] if key.key[-5:] != last_five_chars
+            ]
+            self._bad_jwts.append(last_five_chars)
+
+    def mint_key_for_namespace(
+        self, namespace: str, session_info: dict
+    ) -> DeveloperKey:
+        """
+        Mint a new key for a given namespace
+
+        :param namespace: namespace for the key
+        """
+        salt = secrets.token_hex(32)
+        session_info["salt"] = salt
+        expiry = datetime.utcnow() + timedelta(seconds=self._default_exp)
+        new_key = jwt_encode_user_data(session_info, exp=expiry)
+        key = DeveloperKey(
+            key=new_key,
+            created_at=datetime.utcnow().isoformat(),
+            expires=expiry.isoformat(),
+        )
+        self.add_key(namespace, key)
+        return key
+
+    def is_key_bad(self, last_five_chars: str) -> bool:
+        return last_five_chars in self._bad_jwts
+
+
+dev_key_handler = DeveloperKeyHandler()