ENH: locking down read access on API key

README.md

@@ -124,14 +124,24 @@ Management of these keys is done through a script located at `sds_data_manager/l
 That script can add, remove, and list the current keys. To add keys, add the name and e-mail of the associated
 user or account and get returned an API Key that you can then give to the external user for access.
 
-For example from the command line with the proper AWS credentials:
+##### Scope Options
+
+When creating or updating API keys, you can specify different scopes to control access:
+
+- `full`: Full read and write access to all endpoints and data
+- `read`: Read-only access. Can query and download data but cannot upload or modify files
+
+##### Usage Examples
 
 ```bash
 python sds_data_manager/lambda_code/authorization/manage_api_keys.py list
-python sds_data_manager/lambda_code/authorization/manage_api_keys.py add <owner> <email>
+python sds_data_manager/lambda_code/authorization/manage_api_keys.py add <owner> <email> <scope>
 python sds_data_manager/lambda_code/authorization/manage_api_keys.py remove <key>
 python sds_data_manager/lambda_code/authorization/manage_api_keys.py update_permission <owner> <email> <scope>
 AWS_PROFILE=imap-sdc-dev AWS_DEFAULT_REGION=us-west-2 \
   python sds_data_manager/lambda_code/authorization/manage_api_keys.py \
       add "First Last" "user@example.com" "full"
+AWS_PROFILE=imap-sdc-dev AWS_DEFAULT_REGION=us-west-2 \
+  python sds_data_manager/lambda_code/authorization/manage_api_keys.py \
+      add "Read User" "reader@example.com" "read"
 ```

sds_data_manager/lambda_code/IAlirtCode/ialirt_data_query_api.py

@@ -23,6 +23,11 @@
     "ialirt_scientist",
 }
 
+# Read-only scopes (can read but not write)
+READ_ONLY_SCOPES = {
+    "read",
+}
+
 RESTRICTED_FIELDS = {
     "hit_e_a_side_high_en",
     "hit_e_b_side_high_en",
@@ -169,7 +174,7 @@ def filter_items_by_scope(items: list[dict], scope: str) -> list[dict]:
         Items list.
     """
     # If caller has full HIT access, do nothing
-    if scope in FULL_SCOPES:
+    if scope in FULL_SCOPES or scope in READ_ONLY_SCOPES:
         return items
 
     filtered_items = [

sds_data_manager/lambda_code/SDSCode/api_lambdas/upload_api.py

@@ -105,6 +105,37 @@ def lambda_handler(event, context):
         A pre-signed url where users can upload a data file to the SDS.
 
     """
+    logger.info(f"Received upload request with event: {event}")
+    # Check API key scope for upload restrictions
+    request_ctx = event.get("requestContext", {})
+    auth = request_ctx.get("authorizer", {})
+    auth_ctx = auth.get("lambda", {})
+    scope = auth_ctx.get("scope", "")
+    api_key = auth_ctx.get("apiKey", "unknown")
+
+    logger.info(f"Upload request received with scope: {scope}, api_key: {api_key}")
+
+    # Deny upload access for read scope
+    if scope == "read":
+        logger.warning("Upload denied: read scope user attempted upload")
+        return {
+            "statusCode": 403,
+            "body": json.dumps(
+                "Upload access denied. Your API key has read permissions."
+            ),
+        }
+
+    # Check if scope is missing (might be caught by authorizer first)
+    if not scope:
+        logger.warning("Upload denied: no scope found in authorizer context")
+        return {
+            "statusCode": 403,
+            "body": json.dumps(
+                "Upload access denied. Please provide a valid API key with "
+                "upload permissions."
+            ),
+        }
+
     path_params = event.get("pathParameters", {}).get("proxy", None)
     logger.info("Parsing path parameters=[%s] from event=[%s]", path_params, event)
 

sds_data_manager/lambda_code/authorization/lambda_api_key_authorizer.py

@@ -1,54 +1,114 @@
 """Authorization for API Keys within the SDS."""
 
+import logging
+
 import boto3
 
+# Configure logging
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
 # Initialize DynamoDB resource
 # Specifically outside of the handler to be cached in the lambda execution environment
 dynamodb = boto3.resource("dynamodb")
 table = dynamodb.Table("imap-sdc-api-keys")
 
 
+def _is_authorized(scope, path, http_method):
+    """Check if the API key is authorized for the requested operation.
+
+    Parameters
+    ----------
+    scope : str
+        The scope/permission level of the API key
+    path : str
+        The request path
+    http_method : str
+        The HTTP method (GET, POST, PUT, etc.)
+
+    Returns
+    -------
+    bool
+        True if authorized, False otherwise
+    """
+    logger.info(
+        f"Checking authorization - scope: {scope}, path: {path}, method: {http_method}"
+    )
+
+    # Restrict write operations for read scope
+    if scope == "read" and http_method in ("PUT", "POST", "DELETE", "PATCH"):
+        logger.warning(
+            f"DENIED: read scope user attempted {http_method} operation on {path}"
+        )
+        return False
+
+    # Restrict write operations (upload) for read scope
+    if scope == "read" and path.startswith("/api-key/upload"):
+        logger.warning(f"DENIED: read scope user attempted upload on {path}")
+        return False
+
+    # Public download except for logs and packets.
+    if (
+        path.startswith("/ialirt-download/logs")
+        or path.startswith("/ialirt-download/packets/")
+    ) and scope not in (
+        "full",
+        "read",
+    ):
+        logger.warning(
+            f"DENIED: scope '{scope}' not authorized for I-ALiRT download endpoint"
+        )
+        return False
+
+    logger.info(f"AUTHORIZED: API key with scope '{scope}' granted access to {path}")
+    return True
+
+
 def lambda_handler(event, context):
     """Get the API Key from the request header and check if it is valid."""
+    logger.info(f"Received authorization request with event: {event}")
     api_key = event.get("headers", {}).get("x-api-key", None)
 
     if not api_key:
+        logger.warning("DENIED: No API key provided in request headers")
         return {"isAuthorized": False}
 
+    logger.info("API key received. Checking authorization...")
+
     # Retrieve metadata from DynamoDB
     try:
         metadata = table.get_item(Key={"api_key": api_key}).get("Item")
-    except Exception:
-        # Log? print(f"Error retrieving API key metadata: {e}")
+    except Exception as e:
+        logger.error(f"Error retrieving API key metadata from DynamoDB: {e}")
         return {"isAuthorized": False}
+
     if not metadata:
+        logger.warning("DENIED: API key not found in database")
         return {"isAuthorized": False}
 
     scope = metadata.get("scope", "")
     path = event.get("rawPath") or event.get("path", "")
+    http_method = event.get("requestContext", {}).get("http", {}).get("method", "GET")
 
-    # Check scope-based authorization for specific endpoints
-    if path.startswith("/ialirt-db-query") and scope not in (
-        "ialirt_db",
-        "full",
-        "ialirt_external_partner",
-        "ialirt_scientist",
-    ):
-        return {"isAuthorized": False}
+    logger.info(f"API key found with scope: {scope}")
+    logger.info(f"Request details - Path: {path}, Method: {http_method}")
 
-    # Public download except for logs and packets.
-    if (
-        path.startswith("/ialirt-download/logs")
-        or path.startswith("/ialirt-download/packets/")
-    ) and scope not in (
-        "full",
-        "ialirt_external_partner",
-        "ialirt_scientist",
-    ):
-        return {"isAuthorized": False}
+    is_authorized = _is_authorized(scope, path, http_method)
+    if not is_authorized:
+        logger.warning(
+            f"DENIED: API key with scope '{scope}' is not authorized to upload file"
+        )
+        return {
+            "isAuthorized": False,
+            "context": {
+                "apiKey": api_key,
+                "scope": scope,
+            },
+        }
 
+    logger.info(f"Authorization successful for scope '{scope}'")
     return {
-        "isAuthorized": True,
+        "isAuthorized": is_authorized,
         "context": {
             "apiKey": api_key,
             "scope": scope,

sds_data_manager/lambda_code/authorization/manage_api_keys.py

@@ -24,6 +24,12 @@
 
 TABLE_NAME = "imap-sdc-api-keys"
 
+# Validate scope
+VALID_SCOPES = {
+    "full",
+    "read",
+}
+
 
 def get_table():
     """Get the DynamoDB table resource."""
@@ -106,7 +112,25 @@ def list_keys():
 
 
 def add_key(owner, email, scope="full"):
-    """Generate and add a new API key with owner and email metadata."""
+    """Generate and add a new API key with owner and email metadata.
+
+    Parameters
+    ----------
+    owner : str
+        The owner name of the API key
+    email : str
+        The email associated with the API key
+    scope : str
+        The scope/permission level for the key. Valid values are:
+        - 'full': Full read and write access
+        - 'read': Read-only access
+        Default is 'full'
+    """
+    if scope not in VALID_SCOPES:
+        valid_scopes_str = ", ".join(sorted(VALID_SCOPES))
+        print(f"Error: Invalid scope '{scope}'. Valid scopes are: {valid_scopes_str}")
+        return
+
     keys = get_keys()
     # Generate a secure random 32-byte hex key
     new_key = secrets.token_hex(32)
@@ -116,6 +140,7 @@ def add_key(owner, email, scope="full"):
     created = datetime.now().isoformat()
     add_key_to_db(new_key, owner, email, scope, created)
     print(f"Added key: {new_key}")
+    print(f"Scope: {scope}")
     print("Share this key securely with the user.")
 
 
@@ -130,7 +155,24 @@ def remove_key(key):
 
 
 def update_permission(owner: str, email: str, scope: str):
-    """Update permissions for API key."""
+    """Update permissions for API key.
+
+    Parameters
+    ----------
+    owner : str
+        The owner name of the API key
+    email : str
+        The email associated with the API key
+    scope : str
+        The new scope/permission level. Valid values are:
+        - 'full': Full read and write access
+        - 'read': Read-only access
+    """
+    if scope not in VALID_SCOPES:
+        valid_scopes_str = ", ".join(sorted(VALID_SCOPES))
+        print(f"Error: Invalid scope '{scope}'. Valid scopes are: {valid_scopes_str}")
+        return
+
     table = get_table()
     keys = get_keys()
 
@@ -139,9 +181,9 @@ def update_permission(owner: str, email: str, scope: str):
         for key, value in keys.items()
         if value["owner"] == owner and value["email"] == email
     ]
-    key = keys[matches[0]]
 
     if matches:
+        key = keys[matches[0]]
         table.put_item(
             Item={
                 "api_key": matches[0],
@@ -152,6 +194,7 @@ def update_permission(owner: str, email: str, scope: str):
             }
         )
         print(f"Updated key permission for: {owner}, {email}")
+        print(f"New scope: {scope}")
     else:
         print(
             f"Update not performed since no api key match found for: {owner}, {email}."
@@ -162,24 +205,31 @@ def main():
     """CLI entry point."""
     if len(sys.argv) < 2:
         print(
-            "Usage: python manage_api_keys.py [list|add|remove] "
+            "Usage: python manage_api_keys.py [list|add|remove|update_permission] "
             "<key> [owner] [email] [scope]"
         )
         sys.exit(1)
     cmd = sys.argv[1]
     if cmd == "list":
         list_keys()
     elif cmd == "add" and len(sys.argv) == 5:
-        add_key(sys.argv[2], sys.argv[3], sys.argv[4])
+        owner = sys.argv[2]
+        email = sys.argv[3]
+        scope = sys.argv[4]
+        add_key(owner, email, scope)
     elif cmd == "remove" and len(sys.argv) == 3:
         remove_key(sys.argv[2])
     elif cmd == "update_permission":
-        update_permission(sys.argv[2], sys.argv[3], sys.argv[4])
+        owner = sys.argv[2]
+        email = sys.argv[3]
+        scope = sys.argv[4]
+        update_permission(owner, email, scope)
     else:
-        print(
-            "Usage: python manage_api_keys.py [list|add|remove] "
-            "<key> [owner] [email] [scope]"
-        )
+        print("Usage:")
+        print("  python manage_api_keys.py list")
+        print("  python manage_api_keys.py add <owner> <email> [scope]")
+        print("  python manage_api_keys.py remove <key>")
+        print("  python manage_api_keys.py update_permission <owner> <email> <scope>")
         sys.exit(1)