| from __future__ import annotations |
|
|
| import re |
| from abc import ABC |
| from typing import Any, Dict, List |
|
|
| from cloudenv.server.providers.aws.audit_logger import AwsApiAuditLogger |
| from cloudenv.server.providers.aws.client import get_client |
| from cloudenv.server.providers.aws.iam_enforcer import IAMEnforcer |
|
|
| _PASCAL_TO_SNAKE_OVERRIDES: Dict[str, str] = { |
| "ListWebACLs": "list_web_acls", |
| } |
|
|
|
|
| def pascal_to_snake(name: str) -> str: |
| if name in _PASCAL_TO_SNAKE_OVERRIDES: |
| return _PASCAL_TO_SNAKE_OVERRIDES[name] |
| s1 = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", name) |
| return re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1).lower() |
|
|
|
|
| class BaseService(ABC): |
| _service_name: str = "" |
| _boto_service: str = "" |
|
|
| def __init__(self, enforcer: IAMEnforcer, audit_logger: AwsApiAuditLogger): |
| self.enforcer = enforcer |
| self.audit_logger = audit_logger |
| self.client = get_client(self._boto_service) |
|
|
| def handle( |
| self, action: str, payload: Dict[str, Any], principal: str |
| ) -> Any: |
|
|
| instance_ids = self._infer_instance_ids(payload) |
| resources = self._infer_resources(instance_ids) |
|
|
| try: |
| self.enforcer.assert_allowed( |
| principal, |
| self._service_name, |
| action, |
| resources=resources, |
| instance_ids=instance_ids, |
| ) |
|
|
| result = self._execute(action, payload, instance_ids) |
|
|
| self.audit_logger.emit( |
| {}, |
| service=self._service_name, |
| operation=action, |
| instance_id=instance_ids[0] if instance_ids else None, |
| principal=principal, |
| status="success", |
| result_summary=str(result)[:500], |
| ) |
| return result |
|
|
| except Exception as e: |
| self.audit_logger.emit( |
| {}, |
| service=self._service_name, |
| operation=action, |
| instance_id=instance_ids[0] if instance_ids else None, |
| principal=principal, |
| status="error", |
| error=str(e), |
| ) |
| raise |
|
|
| def _execute( |
| self, action: str, payload: Dict[str, Any], instance_ids: List[str] |
| ) -> Any: |
| fn = pascal_to_snake(action) |
| method = getattr(self.client, fn) |
| return method(**payload) |
|
|
| def _infer_instance_ids(self, payload: Dict[str, Any]) -> List[str]: |
| return [] |
|
|
| def _infer_resources(self, instance_ids: List[str]) -> List[str]: |
| return instance_ids or ["*"] |
|
|
| def reset(self) -> None: |
| pass |
|
|
| def health(self) -> float: |
| return 1.0 |
|
|
| def describe_state(self) -> Dict[str, Any]: |
| return {} |
|
|
