Deployment configuration

src/db.py

@@ -0,0 +1,62 @@
+import mysql.connector
+from mysql.connector import pooling
+import threading
+from hvac_handler import get_secret
+import os
+import time
+import sys
+
+
+db_username = get_secret("secret/api-internal/db", "username")
+db_password = get_secret("secret/api-internal/db", "password")
+db_host = os.getenv("BACKEND_API_INTERNAL_DB_HOST","localhost")
+db_database = os.getenv("BACKEND_API_INTERNAL_DB_DATABASE","app")
+
+MAX_RETRIES = 5
+RETRY_DELAY = 5
+
+MYSQL_CONFIG = {
+	"host": db_host,
+	"user": db_username,
+	"password": db_password,
+	"database": db_database
+}
+
+_pool_lock = threading.Lock()
+_connection_pool = None
+
+def create_connection_pool():
+	global _connection_pool
+	for attempt in range(1, MAX_RETRIES+1):
+		try:
+			print(f"[MySQL] Attempt {attempt} to connect...")
+			_connection_pool = mysql.connector.pooling.MySQLConnectionPool(
+				pool_name="mypool",
+				pool_size=5,
+				pool_reset_session=True,
+				**MYSQL_CONFIG
+			)
+			print("[MySQL] Connection pool created successfully.")
+			return
+		except mysql.connector.Error as e:
+			print(f"[MySQL] Attempt {attempt} failed: {e}")
+			if attempt < MAX_RETRIES:
+				time.sleep(RETRY_DELAY)
+	print(f"[MySQL] Failed to connect after {MAX_RETRIES} attempts — exiting.")
+	sys.exit(1)
+
+def get_connection_pool():
+	global _connection_pool
+	with _pool_lock:
+		if _connection_pool is None:
+			create_connection_pool
+	return _connection_pool
+
+def get_db():
+	pool = get_connection_pool()
+	conn = pool.get_connection()
+	try:
+		yield conn
+	finally:
+		conn.close()
+

src/hvac_handler.py

@@ -0,0 +1,44 @@
+import hvac
+import base64
+import os
+import time
+import sys
+
+HVAC_AGENT_URL = os.getenv("HVAC_AGENT_URL","http://vault-agent:8201")
+
+MAX_RETRIES = 5
+BACKOFF = 5
+
+def get_client():
+	for attempt in range(1, MAX_RETRIES+1):
+		try:
+			client = hvac.Client(url=HVAC_AGENT_URL)
+			if client.is_authenticated():
+				return client
+			raise Exception("Not authenticated")
+		except Exception as e:
+			print(f"Vault connection failed (attempt {attempt}/{MAX_RETRIES}): {e}")
+			time.sleep(BACKOFF * attempt)
+	print("Vault unreachable after retries. Exiting.")
+	sys.exit(1)
+
+client = get_client()
+
+def get_secret(path:str, key:str):
+	try:
+		secret = client.secrets.kv.v2.read_secret_version(
+			mount_point="kv",
+			path=path
+		)
+		return secret["data"]["data"][key]
+	except Exception as e:
+		print(f"Failed to fetch secret '{path}:{key}': {e}")
+		sys.exit(1)
+
+def decrypt_token(ciphertext: str) -> str:
+    response = client.secrets.transit.decrypt_data(
+        name="push-tokens",
+        ciphertext=ciphertext
+    )
+    plaintext_b64 = response["data"]["plaintext"]
+    return base64.b64decode(plaintext_b64).decode()

src/logger_handler.py

@@ -0,0 +1,13 @@
+import logging
+
+def setup_logger(name: str) -> logging.Logger:
+	logger = logging.getLogger(name)
+	if not logger.handlers:
+		handler = logging.StreamHandler()
+		formatter = logging.Formatter(
+			'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+		)
+		handler.setFormatter(formatter)
+		logger.addHandler(handler)
+		logger.setLevel(logging.DEBUG)
+	return logger

src/main.py

@@ -0,0 +1,68 @@
+from fastapi import FastAPI, Depends, HTTPException
+from fastapi.responses import JSONResponse
+from fastapi.security.api_key import APIKeyHeader
+from starlette.exceptions import HTTPException as StarletteHTTPException
+from typing import Dict
+from pydantic import BaseModel
+from validator import verify_api_key
+from db import get_db
+from logger_handler import setup_logger
+from rabbitmq_handler import send_message_to_rmq
+import uvicorn
+from uvicorn_logging_config import LOGGING_CONFIG
+
+logger = setup_logger(__name__)
+
+api_key_header_internal = APIKeyHeader(name="X-API-Key-Internal")
+
+class Notification(BaseModel):
+	receipent_user_id : int
+	message : Dict
+
+api = FastAPI(
+		title="Internal Notifier API",
+		description="API to forward messages to RabbitMQ",
+		version="1.0.0"
+)
+
+def verify_api_key_dependency_internal(db=Depends(get_db), api_key: str = Depends(api_key_header_internal)) -> str:
+	cursor = db.cursor()
+	cursor.execute("SELECT program_name, api_key FROM internal_api_keys WHERE status = 'active'")
+	for program_name, hashed_key in cursor.fetchall():
+		if verify_api_key(api_key=api_key, hashed=hashed_key):
+			return program_name
+	raise HTTPException(status_code=403, detail="Unauthorized")
+
+@api.exception_handler(StarletteHTTPException)
+async def custom_http_exception_handler(request,exc):
+	if exc.status_code == 404:
+		return JSONResponse(
+			status_code=403,
+			content={"detail": "Unauthorized"}
+		)
+	return JSONResponse(
+		status_code=exc.status_code,
+		content={"detail": exc.detail}
+	)
+
+
+@api.post("/internal/receive-notifications")
+def receive_notifications(
+	notification_data: Notification,
+	db = Depends(get_db),
+	program_name: str = Depends(verify_api_key_dependency_internal)
+):
+	logger.info(f"Received notifcation data from {program_name} for RMQ")
+	send_message_to_rmq(notification_data.user_id,notification_data.message)
+	logger.info("Successfully delivered message to RMQ")
+	return {"status": "queued"}
+
+
+if __name__ == "__main__":
+	uvicorn.run(
+		"main:api",
+		host="0.0.0.0",
+		port=8101,
+		log_config=LOGGING_CONFIG,
+		log_level="info"
+	)

src/rabbitmq_handler.py

@@ -0,0 +1,56 @@
+import pika
+from typing import Dict
+import ssl
+from hvac_handler import get_secret
+import json
+import time
+import sys
+
+rmq_username = get_secret("secret/api-internal/rmq", "username")
+rmq_password = get_secret("secret/api-internal/rmq", "password")
+
+MAX_RETRIES = 5
+RETRY_DELAY = 5
+
+def send_message_to_rmq(user_id: int, message: Dict):
+	credentials = pika.PlainCredentials(username=rmq_username, password=rmq_password)
+	context = ssl.create_default_context()
+	context.check_hostname = False
+	ssl_options = pika.SSLOptions(context)
+	conn_params = pika.ConnectionParameters(
+		host="localhost",
+		port=5671,
+		ssl_options=ssl_options,
+		credentials=credentials,
+		virtual_host="app_notifications"
+	)
+
+	for attempt in range(1, MAX_RETRIES + 1):
+		try:
+			connection = pika.BlockingConnection(conn_params)
+			channel = connection.channel()
+			channel.exchange_declare(exchange="app_notifications", exchange_type="topic", durable=True)
+			channel.confirm_delivery()
+			channel.basic_publish(
+				exchange='app_notifications',
+				routing_key=f"notify.user.{user_id}",
+				body=json.dumps(message),
+				properties=pika.BasicProperties(
+					content_type="application/json",
+					delivery_mode=2
+				),
+				mandatory=True
+			)
+			connection.close()
+			return
+
+		except Exception as e:
+			print(f"[RMQ] Attempt {attempt} failed: {e}")
+			if attempt < MAX_RETRIES:
+				time.sleep(RETRY_DELAY)
+			else:
+				print("[RMQ] Failed to connect after maximum retries — exiting.")
+				sys.exit(1)
+
+if __name__ == "__main__":
+	send_message_to_rmq(1, {"type": "notification", "content": "Vault TLS cert reloaded successfully."})

src/uvicorn_logging_config.py

@@ -0,0 +1,39 @@
+LOGGING_CONFIG = {
+	"version": 1,
+	"disable_existing_loggers": False,
+	"formatters": {
+		"default": {
+			"format": "%(asctime)s - %(levelname)s - %(name)s - %(message)s",
+			"datefmt": "%Y-%m-%d %H:%M:%S",
+		}
+	},
+	"handlers": {
+		"default": {
+			"class": "logging.StreamHandler",
+			"formatter": "default",
+			"stream": "ext://sys.stdout"
+		}
+	},
+	"loggers": {
+		"": {  # root logger
+			"handlers": ["default"],
+			"level": "INFO",
+			"propagate": False
+		},
+		"uvicorn": {
+			"handlers": ["default"],
+			"level": "INFO",
+			"propagate": False
+		},
+		"uvicorn.error": {
+			"handlers": ["default"],
+			"level": "INFO",
+			"propagate": False
+		},
+		"uvicorn.access": {
+			"handlers": ["default"],
+			"level": "INFO",
+			"propagate": False
+		}
+	}
+}

src/validator.py

@@ -0,0 +1,20 @@
+from argon2 import PasswordHasher
+
+ph = PasswordHasher()
+
+def hash_api_key(api_key: str) -> str:
+	return ph.hash(api_key)
+
+def verify_api_key(api_key: str, hashed: str) -> bool:
+	try:
+		return ph.verify(hashed, api_key)
+	except Exception:
+		return False
+
+if __name__=="__main__":
+	plain_key = "super-secret-api-key"
+	#hashed_key = hash_api_key(plain_key)
+	hashed_key = '$argon2id$v=19$m=65536,t=3,p=4$vqU+MRafVW1b8AtF+zHb0w$p1J4Gyb0jhlVtKgYyjTITxfU97YaayeS3s3qFFP5sVM'
+
+	print("Hashed API Key:", hashed_key)
+	print("Verification:", verify_api_key(plain_key, hashed_key))