로컬 디비 생성 로직 및 utils, 암호화 부분 생성

.github/workflows/build_release_and_notify.yml

@@ -75,6 +75,8 @@ jobs:
       # 6. PyInstaller를 사용해 파이썬 코드를 실행 파일로 만듭니다.
       - name: Build executable with PyInstaller
         shell: bash
+        env:
+          ENV_AES256_KEY: ${{ secrets.ENV_AES256_KEY }}
         run: poetry run pyinstaller --clean --additional-hooks-dir ./hooks --add-data "app/assets:assets" --onefile --name ${{ env.EXE_NAME }} app/main.py
 
       # 7. 빌드된 실행 파일을 다음 단계(deploy)에서 사용할 수 있도록 아티팩트로 업로드합니다.

app/api/driver_api.py

@@ -2,11 +2,11 @@
 
 from fastapi import APIRouter
 
-from app.core.db_driver_enum import DBTypesEnum
+from app.core.enum.db_driver import DBTypesEnum
 from app.core.exceptions import APIException
 from app.core.status import CommonCode
 from app.schemas.driver_info import DriverInfo
-from app.schemas.response import ResponseMessage
+from app.core.response import ResponseMessage
 from app.services.driver_service import db_driver_info
 
 router = APIRouter()

app/api/test_api.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter
 
-from app.schemas.response import ResponseMessage
+from app.core.response import ResponseMessage
 from app.core.exceptions import APIException
 from app.core.status import CommonCode
 

app/core/db_driver_enum.py → app/core/enum/db_driver.py

@@ -1,4 +1,4 @@
-# app/core/db_driver_enum.py
+# app/core/enum/db_driver.py
 from enum import Enum
 
 

app/core/security.py

@@ -0,0 +1,44 @@
+import os
+import base64
+from Crypto.Cipher import AES
+from Crypto.Util.Padding import pad, unpad
+from Crypto.Random import get_random_bytes
+
+"""
+보안 원칙을 적용한 AES-256 암호화 및 복호화 클래스입니다.
+- 암호화 시 매번 새로운 랜덤 IV를 생성합니다.
+"""
+class AES256:
+    _key = base64.b64decode(os.getenv("ENV_AES256_KEY"))
+
+    @staticmethod
+    def encrypt(text: str) -> str:
+        iv = get_random_bytes(AES.block_size)
+
+        cipher = AES.new(AES256._key, AES.MODE_CBC, iv)
+
+        data_bytes = text.encode('utf-8')
+        padded_bytes = pad(data_bytes, AES.block_size)
+
+        encrypted_bytes = cipher.encrypt(padded_bytes)
+
+        combined_bytes = iv + encrypted_bytes
+        return base64.b64encode(combined_bytes).decode('utf-8')
+
+    @staticmethod
+    def decrypt(cipher_text: str) -> str:
+        """
+        AES-256으로 암호화된 텍스트를 복호화합니다.
+        """
+        combined_bytes = base64.b64decode(cipher_text)
+
+        iv = combined_bytes[:AES.block_size]
+        encrypted_bytes = combined_bytes[AES.block_size:]
+
+        cipher = AES.new(AES256._key, AES.MODE_CBC, iv)
+
+        decrypted_padded_bytes = cipher.decrypt(encrypted_bytes)
+        decrypted_bytes = unpad(decrypted_padded_bytes, AES.block_size)
+
+        return decrypted_bytes.decode('utf-8')
+

app/core/utils.py

@@ -0,0 +1,23 @@
+import uuid
+from pathlib import Path
+
+# 앱 데이터를 저장할 폴더 이름
+APP_DATA_DIR_NAME = ".qgenie"
+
+def get_db_path() -> Path:
+    """
+    사용자 홈 디렉터리 내에 앱 데이터 폴더를 만들고,
+    SQLite DB 파일의 전체 경로를 반환합니다.
+    """
+    home_dir = Path.home()
+    app_data_dir = home_dir / APP_DATA_DIR_NAME
+    app_data_dir.mkdir(exist_ok=True)
+    db_path = app_data_dir / "local_storage.sqlite"
+    return db_path
+
+def generate_uuid() -> str:
+    return uuid.uuid4().hex.upper()
+
+def generate_prefixed_uuid(prefix: str) -> str:
+    return f"{prefix.upper()}-{uuid.uuid4().hex.upper()}"
+

app/db/init_db.py

@@ -0,0 +1,130 @@
+# db/init_db.py
+import sqlite3
+from app.core.utils import get_db_path
+
+"""
+데이터베이스에 연결하고, 애플리케이션에 필요한 테이블이 없으면 생성합니다.
+"""
+def initialize_database():
+
+    db_path = get_db_path()
+    conn = None
+    try:
+        conn = sqlite3.connect(db_path)
+        cursor = conn.cursor()
+        # db_profile 테이블 생성
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS db_profile (
+                id VARCHAR(64) PRIMARY KEY NOT NULL,
+                type VARCHAR(32) NOT NULL,
+                host VARCHAR(255) NOT NULL,
+                port INTEGER NOT NULL,
+                name VARCHAR(64),
+                username VARCHAR(128) NOT NULL,
+                password VARCHAR(128) NOT NULL,
+                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+            );
+        """)
+        # db_profile 테이블의 updated_at을 자동으로 업데이트하는 트리거
+        cursor.execute("""
+            CREATE TRIGGER IF NOT EXISTS update_db_profile_updated_at
+            BEFORE UPDATE ON db_profile
+            FOR EACH ROW
+            BEGIN
+                UPDATE db_profile SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;
+            END;
+        """)
+
+        # ai_credential 테이블 생성
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS ai_credential (
+                id VARCHAR(64) PRIMARY KEY NOT NULL,
+                service_name VARCHAR(32) NOT NULL,
+                api_key VARCHAR(256) NOT NULL,
+                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+            );
+        """)
+        # ai_credential 테이블의 updated_at을 자동으로 업데이트하는 트리거
+        cursor.execute("""
+            CREATE TRIGGER IF NOT EXISTS update_ai_credential_updated_at
+            BEFORE UPDATE ON ai_credential
+            FOR EACH ROW
+            BEGIN
+                UPDATE ai_credential SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;
+            END;
+        """)
+
+        # chat_tab 테이블 생성
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS chat_tab (
+                id VARCHAR(64) PRIMARY KEY NOT NULL,
+                name VARCHAR(255),
+                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+            );
+        """)
+        # chat_tab 테이블의 updated_at을 자동으로 업데이트하는 트리거
+        cursor.execute("""
+            CREATE TRIGGER IF NOT EXISTS update_chat_tab_updated_at
+            BEFORE UPDATE ON chat_tab
+            FOR EACH ROW
+            BEGIN
+                UPDATE chat_tab SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;
+            END;
+        """)
+
+        # chat_message 테이블 생성
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS chat_message (
+                id VARCHAR(64) PRIMARY KEY NOT NULL,
+                chat_tab_id VARCHAR(64) NOT NULL,
+                sender VARCHAR(1) NOT NULL,
+                message TEXT NOT NULL,
+                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (chat_tab_id) REFERENCES chat_tab(id)
+            );
+        """)
+        # chat_message 테이블의 updated_at을 자동으로 업데이트하는 트리거
+        cursor.execute("""
+            CREATE TRIGGER IF NOT EXISTS update_chat_message_updated_at
+            BEFORE UPDATE ON chat_message
+            FOR EACH ROW
+            BEGIN
+                UPDATE chat_message SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;
+            END;
+        """)
+
+        # query_history 테이블 생성
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS query_history (
+                id VARCHAR(64) PRIMARY KEY NOT NULL,
+                chat_message_id VARCHAR(64) NOT NULL,
+                query_text TEXT NOT NULL,
+                is_success VARCHAR(1) NOT NULL,
+                error_message TEXT NOT NULL,
+                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (chat_message_id) REFERENCES chat_message(id)
+            );
+        """)
+        # query_history 테이블의 updated_at을 자동으로 업데이트하는 트리거
+        cursor.execute("""
+            CREATE TRIGGER IF NOT EXISTS update_query_history_updated_at
+            BEFORE UPDATE ON query_history
+            FOR EACH ROW
+            BEGIN
+                UPDATE query_history SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;
+            END;
+        """)
+
+        conn.commit()
+
+    except sqlite3.Error as e:
+        print(f"데이터베이스 초기화 중 오류 발생: {e}")
+    finally:
+        if conn:
+            conn.close()
+

app/main.py

@@ -6,6 +6,7 @@
 from app.api import health  # 헬스 체크
 from app.api.api_router import api_router
 from app.core.exceptions import APIException, api_exception_handler, generic_exception_handler
+from app.db.init_db import initialize_database
 
 app = FastAPI()
 
@@ -17,6 +18,8 @@
 app.include_router(health.router)
 app.include_router(api_router, prefix="/api")
 
+# initialize_database 함수가 호출되어 테이블이 생성되거나 이미 존재함을 확인합니다.
+initialize_database()
 
 if __name__ == "__main__":
     # Uvicorn 서버를 시작합니다.

app/schemas/driver_info.py

@@ -1,7 +1,7 @@
 # app/schemas/driver_info.py
 from pydantic import BaseModel
 
-from app.core.db_driver_enum import DBTypesEnum
+from app.core.enum.db_driver import DBTypesEnum
 
 
 class DriverInfo(BaseModel):

pyproject.toml

@@ -17,7 +17,8 @@ dependencies = [
     "mysql-connector-python (>=9.4.0,<10.0.0)",
     "pymysql (>=1.1.1,<2.0.0)",
     "cx-oracle (>=8.3.0,<9.0.0)",
-    "pyodbc (>=5.2.0,<6.0.0)"
+    "pyodbc (>=5.2.0,<6.0.0)",
+    "pycryptodome (>=3.23.0,<4.0.0)"
 ]