| 1 | from typing import Any |
| 2 | from typing import Dict |
| 3 | from typing import Optional |
| 4 | from urllib.parse import urljoin |
| 5 | |
| 6 | import requests |
| 7 | from fastapi import HTTPException |
| 8 | from loguru import logger |
| 9 | from sqlalchemy.ext.asyncio import AsyncSession |
| 10 | from sqlalchemy.future import select |
| 11 | |
| 12 | from app.connectors.utils import get_connector_info_from_db |
| 13 | from app.db.db_session import get_db_session |
| 14 | from app.db.universal_models import CustomersMeta |
| 15 | |
| 16 | |
| 17 | async def get_endpoint_id() -> int: |
| 18 | """ |
| 19 | Returns the ID of the endpoint. |
| 20 | """ |
| 21 | logger.info("Getting endpoint ID") |
| 22 | list_endpoints = await send_get_request("/api/endpoints") |
| 23 | |
| 24 | for endpoint in list_endpoints["data"]: |
| 25 | if endpoint["Name"] == "local": |
| 26 | # Convert the ID to an integer |
| 27 | return int(endpoint["Id"]) |
| 28 | if endpoint["Name"] == "primary": |
| 29 | return int(endpoint["Id"]) |
| 30 | return None |
| 31 | |
| 32 | |
| 33 | async def get_swarm_id() -> int: |
| 34 | """ |
| 35 | Returns the ID of the swarm. |
| 36 | """ |
| 37 | logger.info("Getting swarm ID") |
| 38 | endpoint_id = await get_endpoint_id() |
| 39 | logger.info(f"Endpoint ID: {endpoint_id}") |
| 40 | swarm_id = await send_get_request(f"/api/endpoints/{endpoint_id}/docker/swarm") |
| 41 | return swarm_id["data"]["ID"] |
| 42 | |
| 43 | |
| 44 | async def get_portainer_jwt() -> str: |
| 45 | """Get JWT token from Portainer API.""" |
| 46 | logger.info("Getting portainer authentication token") |
| 47 | async with get_db_session() as session: # This will correctly enter the context manager |
| 48 | attributes = await get_connector_info_from_db("Portainer", session) |
| 49 | try: |
| 50 | auth_endpoint = urljoin(attributes["connector_url"], "/api/auth") |
| 51 | |
| 52 | auth_payload = {"username": attributes["connector_username"], "password": attributes["connector_password"]} |
| 53 | |
| 54 | response = requests.post(auth_endpoint, json=auth_payload, verify=False) # If using self-signed cert |
| 55 | |
| 56 | response.raise_for_status() |
| 57 | # The JWT token is in response.json()["jwt"] |
| 58 | jwt_token = response.json()["jwt"] |
| 59 | logger.info(f"Authenticated with Portainer. JWT token: {jwt_token}") |
| 60 | return jwt_token |
| 61 | |
| 62 | except requests.exceptions.RequestException as e: |
| 63 | error_msg = f"Failed to authenticate with Portainer: {str(e)}" |
| 64 | logger.error(error_msg) |
| 65 | raise HTTPException(status_code=500, detail=error_msg) |
| 66 | |
| 67 | |
| 68 | async def verify_portainer_credentials(attributes: Dict[str, Any]) -> Dict[str, Any]: |
| 69 | """ |
| 70 | Verifies the connection to portainer service by attempting both API key and JWT authentication. |
| 71 | |
| 72 | Returns: |
| 73 | dict: A dictionary containing 'connectionSuccessful' status and authentication details. |
| 74 | """ |
| 75 | logger.info( |
| 76 | f"Verifying the portainer connection to {attributes['connector_url']}", |
| 77 | ) |
| 78 | try: |
| 79 | # First try API key authentication |
| 80 | headers = { |
| 81 | "Authorization": f"Bearer {attributes['connector_api_key']}", |
| 82 | } |
| 83 | portainer_apps = requests.get( |
| 84 | f"{attributes['connector_url']}/api/v1/apps/authentication", |
| 85 | headers=headers, |
| 86 | verify=False, |
| 87 | timeout=2, |
| 88 | ) |
| 89 | |
| 90 | # If API key auth fails, try JWT authentication |
| 91 | if portainer_apps.status_code != 200: |
| 92 | auth_endpoint = urljoin(attributes["connector_url"], "/api/auth") |
| 93 | auth_payload = {"username": attributes["connector_username"], "password": attributes["connector_password"]} |
| 94 | |
| 95 | jwt_response = requests.post(auth_endpoint, json=auth_payload, verify=False, timeout=2) |
| 96 | |
| 97 | if jwt_response.status_code == 200: |
| 98 | jwt_token = jwt_response.json()["jwt"] |
| 99 | logger.info("JWT authentication successful") |
| 100 | return { |
| 101 | "connectionSuccessful": True, |
| 102 | "message": "Portainer connection successful via JWT", |
| 103 | "authMethod": "jwt", |
| 104 | "jwt": jwt_token, |
| 105 | } |
| 106 | else: |
| 107 | logger.error(f"Both API key and JWT authentication failed. JWT error: {jwt_response.text}") |
| 108 | return {"connectionSuccessful": False, "message": "Both API key and JWT authentication failed", "authMethod": None} |
| 109 | |
| 110 | logger.info( |
| 111 | f"Connection to {attributes['connector_url']} successful via API key", |
| 112 | ) |
| 113 | return {"connectionSuccessful": True, "message": "Portainer connection successful via API key", "authMethod": "api_key"} |
| 114 | |
| 115 | except Exception as e: |
| 116 | logger.error( |
| 117 | f"Connection to {attributes['connector_url']} failed with error: {e}", |
| 118 | ) |
| 119 | return { |
| 120 | "connectionSuccessful": False, |
| 121 | "message": f"Connection to {attributes['connector_url']} failed with error: {e}", |
| 122 | "authMethod": None, |
| 123 | } |
| 124 | |
| 125 | |
| 126 | async def verify_portainer_connection(connector_name: str) -> str: |
| 127 | """ |
| 128 | Returns if connection to portainer service is successful. |
| 129 | """ |
| 130 | logger.info("Getting portainer authentication token") |
| 131 | async with get_db_session() as session: # This will correctly enter the context manager |
| 132 | attributes = await get_connector_info_from_db(connector_name, session) |
| 133 | if attributes is None: |
| 134 | logger.error("No portainer connector found in the database") |
| 135 | return None |
| 136 | return await verify_portainer_credentials(attributes) |
| 137 | |
| 138 | |
| 139 | async def send_get_request( |
| 140 | endpoint: str, |
| 141 | params: Optional[Dict[str, Any]] = None, |
| 142 | connector_name: str = "Portainer", |
| 143 | ) -> Dict[str, Any]: |
| 144 | """ |
| 145 | Sends a GET request to the portainer service. |
| 146 | |
| 147 | Args: |
| 148 | endpoint (str): The endpoint to send the GET request to. |
| 149 | params (Optional[Dict[str, Any]], optional): The parameters to send with the GET request. Defaults to None. |
| 150 | connector_name (str, optional): The name of the connector to use. Defaults to "portainer". |
| 151 | |
| 152 | Returns: |
| 153 | Dict[str, Any]: The response from the GET request. |
| 154 | """ |
| 155 | logger.info(f"Sending GET request to {endpoint}") |
| 156 | async with get_db_session() as session: # This will correctly enter the context manager |
| 157 | attributes = await get_connector_info_from_db(connector_name, session) |
| 158 | if attributes is None: |
| 159 | logger.error("No portainer connector found in the database") |
| 160 | return None |
| 161 | jwt_token = await get_portainer_jwt() |
| 162 | try: |
| 163 | HEADERS = { |
| 164 | "Authorization": f"Bearer {jwt_token}", |
| 165 | "Content-Type": "application/json", |
| 166 | } |
| 167 | logger.info(f"Sending GET request to {attributes['connector_url']}{endpoint}") |
| 168 | response = requests.get( |
| 169 | f"{attributes['connector_url']}{endpoint}", |
| 170 | headers=HEADERS, |
| 171 | params=params, |
| 172 | verify=False, |
| 173 | ) |
| 174 | return { |
| 175 | "data": response.json(), |
| 176 | "success": True, |
| 177 | "message": "Successfully retrieved data", |
| 178 | } |
| 179 | except Exception as e: |
| 180 | logger.error(f"Failed to send GET request to {endpoint} with error: {e}") |
| 181 | raise HTTPException( |
| 182 | status_code=500, |
| 183 | detail=f"Failed to send GET request to {endpoint} with error: {e}", |
| 184 | ) |
| 185 | return { |
| 186 | "success": False, |
| 187 | "message": f"Failed to send GET request to {endpoint} with error: {e}", |
| 188 | } |
| 189 | |
| 190 | |
| 191 | async def send_post_request( |
| 192 | endpoint: str, |
| 193 | data: Dict[str, Any] = None, |
| 194 | connector_name: str = "Portainer", |
| 195 | ) -> Dict[str, Any]: |
| 196 | """ |
| 197 | Sends a POST request to the portainer service. |
| 198 | |
| 199 | Args: |
| 200 | endpoint (str): The endpoint to send the POST request to. |
| 201 | data (Dict[str, Any]): The data to send with the POST request. |
| 202 | connector_name (str, optional): The name of the connector to use. Defaults to "portainer". |
| 203 | |
| 204 | Returns: |
| 205 | Dict[str, Any]: The response from the POST request. |
| 206 | """ |
| 207 | logger.info(f"Sending POST request to {endpoint}") |
| 208 | async with get_db_session() as session: # This will correctly enter the context manager |
| 209 | attributes = await get_connector_info_from_db(connector_name, session) |
| 210 | if attributes is None: |
| 211 | logger.error("No portainer connector found in the database") |
| 212 | return None |
| 213 | jwt_token = await get_portainer_jwt() |
| 214 | |
| 215 | try: |
| 216 | HEADERS = { |
| 217 | "Authorization": f"Bearer {jwt_token}", |
| 218 | "Content-Type": "application/json", |
| 219 | } |
| 220 | logger.info(f"Sending POST request to {attributes['connector_url']}{endpoint}") |
| 221 | response = requests.post( |
| 222 | f"{attributes['connector_url']}{endpoint}", |
| 223 | headers=HEADERS, |
| 224 | json=data, |
| 225 | verify=False, |
| 226 | ) |
| 227 | |
| 228 | if response.status_code == 204: |
| 229 | return { |
| 230 | "data": None, |
| 231 | "success": True, |
| 232 | "message": "Successfully completed request with no content", |
| 233 | } |
| 234 | else: |
| 235 | return { |
| 236 | "data": response.json(), |
| 237 | "success": False if response.status_code >= 400 else True, |
| 238 | "message": "Successfully retrieved data" if response.status_code < 400 else "Failed to retrieve data", |
| 239 | } |
| 240 | except Exception as e: |
| 241 | logger.debug(f"Response: {response}") |
| 242 | logger.error(f"Failed to send POST request to {endpoint} with error: {e}") |
| 243 | raise HTTPException( |
| 244 | status_code=500, |
| 245 | detail=f"Failed to send POST request to {endpoint} with error: {e}", |
| 246 | ) |
| 247 | return { |
| 248 | "success": False, |
| 249 | "message": f"Failed to send POST request to {endpoint} with error: {e}", |
| 250 | } |
| 251 | |
| 252 | |
| 253 | async def send_delete_request( |
| 254 | endpoint: str, |
| 255 | params: Optional[Dict[str, Any]] = None, |
| 256 | connector_name: str = "Portainer", |
| 257 | ) -> Dict[str, Any]: |
| 258 | """ |
| 259 | Sends a DELETE request to the Portainer service. |
| 260 | |
| 261 | Args: |
| 262 | endpoint (str): The endpoint to send the DELETE request to. |
| 263 | params (Optional[Dict[str, Any]], optional): The parameters to send with the DELETE request. Defaults to None. |
| 264 | connector_name (str, optional): The name of the connector to use. Defaults to "Portainer". |
| 265 | |
| 266 | Returns: |
| 267 | Dict[str, Any]: The response from the DELETE request. |
| 268 | """ |
| 269 | logger.info(f"Sending DELETE request to {endpoint}") |
| 270 | async with get_db_session() as session: |
| 271 | attributes = await get_connector_info_from_db(connector_name, session) |
| 272 | if attributes is None: |
| 273 | logger.error("No portainer connector found in the database") |
| 274 | return None |
| 275 | |
| 276 | jwt_token = await get_portainer_jwt() |
| 277 | try: |
| 278 | HEADERS = { |
| 279 | "Authorization": f"Bearer {jwt_token}", |
| 280 | "Content-Type": "application/json", |
| 281 | } |
| 282 | response = requests.delete( |
| 283 | f"{attributes['connector_url']}{endpoint}", |
| 284 | headers=HEADERS, |
| 285 | params=params, |
| 286 | verify=False, |
| 287 | ) |
| 288 | |
| 289 | # Check if response is empty or not JSON |
| 290 | if response.status_code == 204 or not response.text.strip(): |
| 291 | return { |
| 292 | "data": None, |
| 293 | "success": True, |
| 294 | "message": "Successfully deleted resource", |
| 295 | } |
| 296 | |
| 297 | try: |
| 298 | return { |
| 299 | "data": response.json(), |
| 300 | "success": True, |
| 301 | "message": "Successfully retrieved data", |
| 302 | } |
| 303 | except ValueError: |
| 304 | # Response is not JSON |
| 305 | return { |
| 306 | "data": None, |
| 307 | "success": True if response.status_code < 400 else False, |
| 308 | "message": f"Delete operation completed with status code {response.status_code}", |
| 309 | } |
| 310 | |
| 311 | except Exception as e: |
| 312 | logger.error(f"Failed to send DELETE request to {endpoint} with error: {e}") |
| 313 | raise HTTPException( |
| 314 | status_code=500, |
| 315 | detail=f"Failed to send DELETE request to {endpoint} with error: {e}", |
| 316 | ) |
| 317 | |
| 318 | |
| 319 | def send_put_request( |
| 320 | endpoint: str, |
| 321 | data: Optional[Dict[str, Any]] = None, |
| 322 | connector_name: str = "portainer", |
| 323 | ) -> Dict[str, Any]: |
| 324 | """ |
| 325 | Sends a PUT request to the portainer service. |
| 326 | |
| 327 | Args: |
| 328 | endpoint (str): The endpoint to send the PUT request to. |
| 329 | data (Optional[Dict[str, Any]]): The data to send with the PUT request. |
| 330 | connector_name (str, optional): The name of the connector to use. Defaults to "portainer". |
| 331 | |
| 332 | Returns: |
| 333 | Dict[str, Any]: The response from the PUT request. |
| 334 | """ |
| 335 | logger.info(f"Sending PUT request to {endpoint}") |
| 336 | attributes = get_connector_info_from_db(connector_name) |
| 337 | if attributes is None: |
| 338 | logger.error("No portainer connector found in the database") |
| 339 | return None |
| 340 | try: |
| 341 | HEADERS = { |
| 342 | "Authorization": f"Bearer {attributes['connector_api_key']}", |
| 343 | } |
| 344 | response = requests.put( |
| 345 | f"{attributes['connector_url']}{endpoint}", |
| 346 | headers=HEADERS, |
| 347 | auth=( |
| 348 | attributes["connector_username"], |
| 349 | attributes["connector_password"], |
| 350 | ), |
| 351 | json=data, |
| 352 | verify=False, |
| 353 | ) |
| 354 | return { |
| 355 | "data": response.json(), |
| 356 | "success": True, |
| 357 | "message": "Successfully retrieved data", |
| 358 | } |
| 359 | except Exception as e: |
| 360 | logger.error(f"Failed to send PUT request to {endpoint} with error: {e}") |
| 361 | raise HTTPException( |
| 362 | status_code=500, |
| 363 | detail=f"Failed to send PUT request to {endpoint} with error: {e}", |
| 364 | ) |
| 365 | |
| 366 | |
| 367 | async def get_customer_portainer_stack_id(customer_name: str, session: AsyncSession) -> int: |
| 368 | """ |
| 369 | Get the Portainer stack ID for a customer. |
| 370 | |
| 371 | Args: |
| 372 | customer_name (str): The name of the customer. |
| 373 | session (AsyncSession): The database session. |
| 374 | |
| 375 | Returns: |
| 376 | int: The Portainer stack ID. |
| 377 | |
| 378 | Raises: |
| 379 | HTTPException: If the customer is not found or has no stack ID |
| 380 | """ |
| 381 | logger.info(f"Getting Portainer stack ID for customer {customer_name}") |
| 382 | |
| 383 | # Get customer metadata |
| 384 | stmt = select(CustomersMeta).where(CustomersMeta.customer_name == customer_name) |
| 385 | result = await session.execute(stmt) |
| 386 | customer_meta = result.scalar_one_or_none() |
| 387 | |
| 388 | if customer_meta is None: |
| 389 | logger.error(f"Customer {customer_name} not found in database") |
| 390 | raise HTTPException(status_code=404, detail=f"Customer {customer_name} not found in database") |
| 391 | |
| 392 | if not customer_meta.customer_meta_portainer_stack_id: |
| 393 | logger.error(f"No Portainer stack ID found for customer {customer_name}") |
| 394 | raise HTTPException(status_code=404, detail=f"No Portainer stack ID found for customer {customer_name}") |
| 395 | |
| 396 | logger.info(f"Found Portainer stack ID {customer_meta.customer_meta_portainer_stack_id} for customer {customer_name}") |
| 397 | return customer_meta.customer_meta_portainer_stack_id |
| 398 | |
| 399 | |
| 400 | async def update_customer_portainer_stack_id(customer_name: str, stack_id: int, session: AsyncSession) -> None: |
| 401 | """ |
| 402 | Update the Portainer stack ID for a customer. |
| 403 | |
| 404 | Args: |
| 405 | customer_name (str): The name of the customer. |
| 406 | stack_id (int): The Portainer stack ID. |
| 407 | session (AsyncSession): The database session. |
| 408 | """ |
| 409 | logger.info(f"Updating Portainer stack ID for customer {customer_name} to {stack_id}") |
| 410 | |
| 411 | # Get customer metadata |
| 412 | stmt = select(CustomersMeta).where(CustomersMeta.customer_name == customer_name) |
| 413 | result = await session.execute(stmt) |
| 414 | customer_meta = result.scalar_one_or_none() |
| 415 | |
| 416 | if customer_meta is None: |
| 417 | logger.error(f"Customer {customer_name} not found in database") |
| 418 | raise HTTPException(status_code=404, detail=f"Customer {customer_name} not found in database") |
| 419 | |
| 420 | customer_meta.customer_meta_portainer_stack_id = stack_id |
| 421 | await session.commit() |
| 422 | |
| 423 | logger.info(f"Updated Portainer stack ID for customer {customer_name} to {stack_id}") |
| 424 | return stack_id |