Secrets verification

.gitignore

@@ -13,3 +13,5 @@ testing-out/
 # Misc
 .vscode/
 
+# Tests
+.coverage

api/pyproject.toml

@@ -76,7 +76,8 @@ addopts = [
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "session"
 filterwarnings = [
-    "ignore:cannot collect test class 'Testing':pytest.PytestCollectionWarning"
+    "ignore:cannot collect test class 'Testing':pytest.PytestCollectionWarning",
+    "ignore::DeprecationWarning:botocore.auth",
 ]
 
 # Register markers to easily select unit or integration tests.

api/requirements.txt

@@ -21,4 +21,7 @@ argon2-cffi>=23.1.0
 # Async Jobs
 arq>=0.26.3
 aiofiles>=24.1.0
-tenacity>=9.1.2
+tenacity>=9.1.2
+
+# Provider SDKs
+boto3>=1.38.0

api/src/app/api/v1/users.py

@@ -1,25 +1,25 @@
 import base64
-from datetime import UTC, datetime
 
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.responses import JSONResponse
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio.session import AsyncSession
 
+from src.app.cloud.creds_factory import CredsFactory
+
 from ...core.auth.auth import get_current_user
 from ...core.config import settings
 from ...core.db.database import async_get_db
+from ...crud.crud_secrets import get_user_secrets, upsert_user_secrets
 from ...crud.crud_users import get_user_by_id, update_user_password
 from ...models.secret_model import SecretModel
 from ...models.user_model import UserModel
 from ...schemas.message_schema import (
-    AWSUpdateSecretMessageSchema,
-    AzureUpdateSecretMessageSchema,
+    MessageSchema,
     UpdatePasswordMessageSchema,
 )
 from ...schemas.secret_schema import (
-    AWSSecrets,
-    AzureSecrets,
+    AnySecrets,
     CloudSecretStatusSchema,
     UserSecretResponseSchema,
 )
@@ -116,7 +116,7 @@ async def update_password(
 
 
 @router.get("/me/secrets")
-async def get_user_secrets(
+async def fetch_user_secrets(
     current_user: UserModel = Depends(get_current_user),  # noqa: B008
     db: AsyncSession = Depends(async_get_db),  # noqa: B008
 ) -> UserSecretResponseSchema:
@@ -170,115 +170,67 @@ async def get_user_secrets(
     )
 
 
-@router.post("/me/secrets/aws")
-async def update_aws_secrets(
-    aws_secrets: AWSSecrets,
+@router.post("/me/secrets")
+async def update_user_secrets(
+    creds: AnySecrets,
     current_user: UserModel = Depends(get_current_user),  # noqa: B008
     db: AsyncSession = Depends(async_get_db),  # noqa: B008
-) -> AWSUpdateSecretMessageSchema:
-    """Update the current user's AWS secrets.
+) -> MessageSchema:
+    """Update the current user's secrets.
 
     Args:
     ----
-        aws_secrets (AWSSecrets): The AWS credentials to store.
+        creds (AnySecrets): The provider credentials to store.
         current_user (UserModel): The authenticated user.
         db (Session): Database connection.
 
     Returns:
     -------
-        AWSUpdateSecretMessageSchema: Status message.
+        MessageSchema: Status message of updating user secrets.
 
     """
-    # Fetch secrets explicitly from the database
-    stmt = select(SecretModel).where(SecretModel.user_id == current_user.id)
-    result = await db.execute(stmt)
-    secrets = result.scalars().first()
+    # Fetch secrets from the database
+    secrets = await get_user_secrets(db, current_user.id)
     if not secrets:
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="User secrets record not found",
-        )
-
-    # Encrypt the AWS credentials using the user's public key
-    if not current_user.public_key:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="User encryption keys not set up. Please register a new account.",
+            detail="User secrets record not found!",
         )
 
-    # Convert to dictionary for encryption
-    aws_data = {
-        "aws_access_key": aws_secrets.aws_access_key,
-        "aws_secret_key": aws_secrets.aws_secret_key,
-    }
-
-    # Encrypt with the user's public key
-    encrypted_data = encrypt_with_public_key(aws_data, current_user.public_key)
-
-    # Update the secrets with encrypted values
-    secrets.aws_access_key = encrypted_data["aws_access_key"]
-    secrets.aws_secret_key = encrypted_data["aws_secret_key"]
-    secrets.aws_created_at = datetime.now(UTC)
-    await db.commit()
-
-    return AWSUpdateSecretMessageSchema(message="AWS credentials updated successfully")
-
-
-@router.post("/me/secrets/azure")
-async def update_azure_secrets(
-    azure_secrets: AzureSecrets,
-    current_user: UserModel = Depends(get_current_user),  # noqa: B008
-    db: AsyncSession = Depends(async_get_db),  # noqa: B008
-) -> AzureUpdateSecretMessageSchema:
-    """Update the current user's Azure secrets.
-
-    Args:
-    ----
-        azure_secrets (AzureSecrets): The Azure credentials to store.
-        current_user (UserModel): The authenticated user.
-        db (Session): Database connection.
+    # Verify credentials are valid before storing
+    creds_obj = CredsFactory.create_creds_verification(credentials=creds)
 
-    Returns:
-    -------
-        AzureUpdateSecretMessageSchema: Success message.
+    verified, msg = creds_obj.verify_creds()
 
-    """
-    # Fetch secrets explicitly from the database
-    stmt = select(SecretModel).where(SecretModel.user_id == current_user.id)
-    result = await db.execute(stmt)
-    secrets = result.scalars().first()
-    if not secrets:
+    if not verified:
         raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="User secrets record not found",
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=msg.message,
         )
 
-    # Encrypt the Azure credentials using the user's public key
+    # Encrypt the credentials using the user's public key
     if not current_user.public_key:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail="User encryption keys not set up. Please register a new account.",
         )
 
-    # Convert to dictionary for encryption
-    azure_data = {
-        "azure_client_id": azure_secrets.azure_client_id,
-        "azure_client_secret": azure_secrets.azure_client_secret,
-        "azure_tenant_id": azure_secrets.azure_tenant_id,
-        "azure_subscription_id": azure_secrets.azure_subscription_id,
-    }
+    # Convert provided credentials to dictionary for encryption
+    user_creds = creds_obj.get_user_creds()
 
     # Encrypt with the user's public key
-    encrypted_data = encrypt_with_public_key(azure_data, current_user.public_key)
+    encrypted_data = encrypt_with_public_key(
+        data=user_creds, public_key_b64=current_user.public_key
+    )
 
     # Update the secrets with encrypted values
-    secrets.azure_client_id = encrypted_data["azure_client_id"]
-    secrets.azure_client_secret = encrypted_data["azure_client_secret"]
-    secrets.azure_tenant_id = encrypted_data["azure_tenant_id"]
-    secrets.azure_subscription_id = encrypted_data["azure_subscription_id"]
-    secrets.azure_created_at = datetime.now(UTC)
-    await db.commit()
-
-    return AzureUpdateSecretMessageSchema(
-        message="Azure credentials updated successfully"
+    secrets = creds_obj.update_secret_schema(
+        secrets=secrets, encrypted_data=encrypted_data
+    )
+
+    # Add new secrets to database
+    await upsert_user_secrets(db, secrets, current_user.id)
+
+    return MessageSchema(
+        message=f"{creds.provider.value.upper()} credentials successfully verified and updated."
     )

api/src/app/cloud/__init__.py

@@ -0,0 +1 @@
+"""Cloud controls for the OpenLabs API."""

api/src/app/cloud/aws_creds.py

@@ -0,0 +1,163 @@
+import logging
+from datetime import UTC, datetime
+from typing import List, Tuple
+
+import boto3
+from botocore.exceptions import ClientError
+
+from src.app.schemas.message_schema import MessageSchema
+from src.app.schemas.secret_schema import AWSSecrets, SecretSchema
+
+from .base_creds import AbstractBaseCreds
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+
+class AWSCreds(AbstractBaseCreds):
+    """Credential verification for AWS."""
+
+    credentials: AWSSecrets
+
+    def __init__(self, credentials: AWSSecrets) -> None:
+        """Initialize AWS credentials verification object."""
+        self.credentials = credentials
+
+    def get_user_creds(self) -> dict[str, str]:
+        """Convert user AWS secrets to dictionary for encryption."""
+        return {
+            "aws_access_key": self.credentials.aws_access_key,
+            "aws_secret_key": self.credentials.aws_secret_key,
+        }
+
+    def update_secret_schema(
+        self, secrets: SecretSchema, encrypted_data: dict[str, str]
+    ) -> SecretSchema:
+        """Update user secrets schema with newly encrypted secrets."""
+        secrets.aws_access_key = encrypted_data["aws_access_key"]
+        secrets.aws_secret_key = encrypted_data["aws_secret_key"]
+        secrets.aws_created_at = datetime.now(UTC)
+        return secrets
+
+    def verify_creds(self) -> Tuple[bool, MessageSchema]:
+        """Verify credentials authenticate to an AWS account."""
+        try:
+            # --- Step 1: Basic Authentication with STS ---
+            # Created shared session for authentication and IAM permission check
+            session = boto3.Session(
+                aws_access_key_id=self.credentials.aws_access_key,
+                aws_secret_access_key=self.credentials.aws_secret_key,
+            )
+            client = session.client("sts")
+            caller_identity = (
+                client.get_caller_identity()
+            )  # will raise an error if not valid
+            caller_arn = caller_identity["Arn"]
+            logger.info(
+                "AWS credentials successfully authenticated for ARN: %s", caller_arn
+            )
+
+            if caller_arn.endswith(
+                ":root"
+            ):  # If root access key credentials are used, skip permissions check as root user has all permissions
+                return (
+                    True,
+                    MessageSchema(
+                        message="AWS credentials authenticated and all required permissions are present."
+                    ),
+                )
+
+            # --- Step 2: Simulate permissions for a sample of minimum critical actions ---
+            iam_client = session.client("iam")
+
+            actions_to_test = [
+                # For Instance
+                "ec2:RunInstances",
+                "ec2:TerminateInstances",
+                "ec2:DescribeInstances",
+                # For Vpc
+                "ec2:CreateVpc",
+                "ec2:DeleteVpc",
+                "ec2:DescribeVpcs",
+                # For Subnet
+                "ec2:CreateSubnet",
+                "ec2:DeleteSubnet",
+                "ec2:DescribeSubnets",
+                # For InternetGateway
+                "ec2:CreateInternetGateway",
+                "ec2:DeleteInternetGateway",
+                "ec2:AttachInternetGateway",
+                "ec2:DetachInternetGateway",
+                # For Eip and NatGateway
+                "ec2:AllocateAddress",  # Create EIP
+                "ec2:ReleaseAddress",  # Delete EIP
+                "ec2:AssociateAddress",
+                "ec2:CreateNatGateway",
+                "ec2:DeleteNatGateway",
+                # For KeyPair
+                "ec2:CreateKeyPair",
+                "ec2:DeleteKeyPair",
+                # For SecurityGroup and SecurityGroupRule
+                "ec2:CreateSecurityGroup",
+                "ec2:DeleteSecurityGroup",
+                "ec2:AuthorizeSecurityGroupIngress",
+                "ec2:RevokeSecurityGroupIngress",
+                "ec2:AuthorizeSecurityGroupEgress",
+                "ec2:RevokeSecurityGroupEgress",
+                # For RouteTable, Route, and RouteTableAssociation
+                "ec2:CreateRouteTable",
+                "ec2:DeleteRouteTable",
+                "ec2:CreateRoute",
+                "ec2:DeleteRoute",
+                "ec2:AssociateRouteTable",
+                "ec2:DisassociateRouteTable",
+                # For Transit Gateway ---
+                "ec2:CreateTransitGateway",
+                "ec2:DeleteTransitGateway",
+                "ec2:CreateTransitGatewayVpcAttachment",
+                "ec2:DeleteTransitGatewayVpcAttachment",
+                "ec2:CreateTransitGatewayRoute",
+                "ec2:DeleteTransitGatewayRoute",
+                "ec2:DescribeTransitGateways",
+                "ec2:DescribeTransitGatewayVpcAttachments",
+            ]
+
+            simulation_results = iam_client.simulate_principal_policy(
+                PolicySourceArn=caller_arn, ActionNames=actions_to_test
+            )
+
+            # --- Step 3: Evaluate the simulation results ---
+            denied_actions: List[str] = []
+            for result in simulation_results["EvaluationResults"]:
+                if result["EvalDecision"] != "allowed":
+                    denied_actions.append(result["EvalActionName"])
+
+            if denied_actions:
+                error_message = f"Authentication succeeded, but the user/group is missing required permissions. The following actions were denied: {', '.join(denied_actions)}"
+                logger.error(error_message)
+                return (
+                    False,
+                    MessageSchema(
+                        message=f"Insufficient permissions for your AWS account user/group. Please ensure the following permissions are added: {', '.join(denied_actions)}"
+                    ),
+                )
+            logger.info("All simulated actions were allowed for ARN: %s", caller_arn)
+            return (
+                True,
+                MessageSchema(
+                    message="AWS credentials authenticated and all required permissions are present."
+                ),
+            )
+        except ClientError as e:
+            error_code = e.response["Error"]["Code"]
+            if error_code in ("InvalidClientTokenId", "SignatureDoesNotMatch"):
+                message = "AWS credentials could not be authenticated. Please ensure you are providing credentials that are linked to a valid AWS account."
+            elif error_code == "AccessDenied":
+                message = "AWS credentials are valid, but lack permissions to perform the permissions verification. Please ensure you give your AWS account user/group has the iam:SimulatePrincipalPolicy permission attached to a policy."
+            else:
+                message = e.response["Error"]["Message"]
+            logger.error("AWS verification failed: %s", message)
+            return (
+                False,
+                MessageSchema(message=message),
+            )