Update 2025-12-17 20:47:56

.gitea/workflows/deploy.yml

@@ -10,27 +10,19 @@ jobs:
   test:
     name: Test Build
     runs-on: ubuntu-latest
-
-    # Node 포함 컨테이너 (act_runner + Gitea Actions 필수)
     container: node:18-bullseye
-
     steps:
-      # 1️⃣ 내부 네트워크로 직접 Clone (checkout 액션 제거)
       - name: Checkout repository (internal)
         run: |
           echo "📥 Internal git clone start"
           git clone --depth 1 https://gitea.mouse84.com/Kim.KANGHEE/vconnect-api.git .
           echo "📥 Clone done"
 
-      # 2️⃣ Python 설치
       - name: Install Python
         run: |
           apt-get update
           apt-get install -y python3 python3-pip
-          python3 --version
-          pip3 --version
 
-      # 3️⃣ 의존성 설치
       - name: Install dependencies
         run: |
           pip3 install --upgrade pip
@@ -38,55 +30,61 @@ jobs:
             pip3 install -r requirements.txt
           fi
 
-      # 4️⃣ 기본 테스트 (지금은 echo, 이후 pytest 등으로 교체 가능)
       - name: Run basic tests
         run: |
-          echo "✅ Code checkout success"
-          echo "✅ Python ready"
-          echo "✅ Dependencies installed"
+          echo "✅ Basic tests passed"
 
   deploy:
     name: Deploy to Server
     runs-on: ubuntu-latest
-
     needs: test
     if: github.ref == 'refs/heads/main'
-
     steps:
       - name: Deploy via SSH
         run: |
           set -e
-
           echo "🔐 SSH key setup"
           mkdir -p ~/.ssh
           echo "${{ secrets.SSH_PRIVATE_KEY }}" > ~/.ssh/id_rsa
           chmod 600 ~/.ssh/id_rsa
           ssh-keyscan -H 192.168.0.97 >> ~/.ssh/known_hosts
-
+          
           echo "🚀 Deploy start"
           ssh kdesk84@192.168.0.97 << 'EOF'
             set -e
-
             echo "📂 Move to project directory"
             cd /data/vconnect-api
-
-            echo "📦 Git pull"
-            git pull origin main
-
+            
+            echo "🔄 Git Force Sync"
+            git fetch --all
+            git reset --hard origin/main
+            
+            echo "🐍 Checking Virtual Environment..."
+            # [중요 수정] venv/bin/activate 파일이 없으면 가상환경을 새로 생성함
+            if [ ! -f "venv/bin/activate" ]; then
+              echo "⚠️ venv not found or broken. Creating new virtual environment..."
+              # Ubuntu 24.04에서는 python3-venv 패키지가 필요할 수 있음 (없으면 에러 날 수 있으니 아래 참고)
+              python3 -m venv venv
+              echo "✅ venv created."
+            fi
+            
             echo "🐍 Activate virtualenv"
             source venv/bin/activate
 
+            echo "🔄 stop service"
+            sudo systemctl stop vconnect-api            
+            
             echo "📦 Install dependencies"
             pip install -r requirements.txt
-
+            
             echo "🔄 Restart service"
             sudo systemctl restart vconnect-api
-
+            
             echo "📋 Service status"
             sudo systemctl status vconnect-api --no-pager -l
-
+            
             echo "✅ Deploy finished"
           EOF
-
+          
       - name: Deployment Complete
         run: echo "🎉 VConnect API deployment completed!"

alembic.ini

@@ -0,0 +1,102 @@
+# A generic, single database configuration.
+
+[alembic]
+# path to migration scripts
+script_location = alembic
+
+# template used to generate migration file
+# file_template = %%(rev)s_%%(slug)s
+
+# sys.path path, will be prepended to sys.path if present.
+# defaults to the current working directory.
+prepend_sys_path = .
+
+# timezone to use when rendering the date within the migration file
+# as well as the filename.
+# If specified, requires the python-dateutil library to be installed.
+# string value is passed to dateutil.tz.gettz()
+# leave blank for local time
+# timezone =
+
+# max length of characters to apply to the
+# "slug" field
+# truncate_slug_length = 40
+
+# set to 'true' to run the environment during
+# the 'revision' command, regardless of autogenerate
+# revision_environment = false
+
+# set to 'true' to allow .pyc and .pyo files without
+# a source .py file to be detected as revisions in the
+# versions/ directory
+# sourceless = false
+
+# version location specification; this defaults
+# to alembic/versions.  When using multiple version
+# directories, initial revisions must be specified with --version-path.
+# The path separator used here should be the separator specified by "version_path_separator"
+# below.
+# version_locations = %(here)s/bar:%(here)s/bat:alembic/versions
+
+# version path separator; As mentioned above, this is the character used to split
+# version_locations. The default within new alembic.ini files is "os", which uses os.pathsep.
+# If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas.
+# Valid values for version_path_separator are:
+#
+# version_path_separator = :
+# version_path_separator = ;
+# version_path_separator = space
+version_path_separator = os  # Use os.pathsep. Default configuration used for new projects.
+
+# the output encoding used when revision files
+# are written from script.py.mako
+# output_encoding = utf-8
+
+sqlalchemy.url = driver://user:pass@localhost/dbname
+
+
+[post_write_hooks]
+# post_write_hooks defines scripts or Python functions that are run
+# on newly generated revision scripts.  See the documentation for further
+# detail and examples
+
+# format using "black" - use the console_scripts runner, against the "black" entrypoint
+# hooks = black
+# black.type = console_scripts
+# black.entrypoint = black
+# black.options = -l 79 REVISION_SCRIPT_FILENAME
+
+# Logging configuration
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+qualname =
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S

alembic/env.py

@@ -0,0 +1,88 @@
+from logging.config import fileConfig
+
+from sqlalchemy import engine_from_config
+from sqlalchemy import pool
+
+from alembic import context
+
+# Import your models and config
+from app.config import settings
+from app.database import Base
+# Import all models so Base.metadata has them
+from app.models import user, vm, audit_log 
+
+# this is the Alembic Config object, which provides
+# access to the values within the .ini file in use.
+config = context.config
+
+# Interpret the config file for Python logging.
+# This line sets up loggers basically.
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+# add your model's MetaData object here
+# for 'autogenerate' support
+# from myapp import mymodel
+# target_metadata = mymodel.Base.metadata
+target_metadata = Base.metadata
+
+# other values from the config, defined by the needs of env.py,
+# can be acquired:
+# my_important_option = config.get_main_option("my_important_option")
+# ... etc.
+
+
+def run_migrations_offline() -> None:
+    """Run migrations in 'offline' mode.
+
+    This configures the context with just a URL
+    and not an Engine, though an Engine is acceptable
+    here as well.  By skipping the Engine creation
+    we don't even need a DBAPI to be available.
+
+    Calls to context.execute() here emit the given string to the
+    script output.
+
+    """
+    url = settings.DATABASE_URL
+    context.configure(
+        url=url,
+        target_metadata=target_metadata,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+    )
+
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    """Run migrations in 'online' mode.
+
+    In this scenario we need to create an Engine
+    and associate a connection with the context.
+
+    """
+    # Overwrite the sqlalchemy.url from settings
+    configuration = config.get_section(config.config_ini_section)
+    configuration["sqlalchemy.url"] = settings.DATABASE_URL
+    
+    connectable = engine_from_config(
+        configuration,
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+
+    with connectable.connect() as connection:
+        context.configure(
+            connection=connection, target_metadata=target_metadata
+        )
+
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

alembic/script.py.mako

@@ -0,0 +1,26 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+# revision identifiers, used by Alembic.
+revision: str = ${repr(up_revision)}
+down_revision: Union[str, None] = ${repr(down_revision)}
+branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
+depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}

alembic/versions/.keep

@@ -0,0 +1 @@
+Keep

app/api/ssh_credentials.py

@@ -4,6 +4,7 @@ from datetime import datetime, timedelta
 from app.api.auth import get_current_user
 from app.schemas.auth import CurrentUser
 from app.services.temp_ssh_password_service import temp_ssh_password_manager
+from app.config import settings
 import os
 
 router = APIRouter()
@@ -21,15 +22,38 @@ async def get_ssh_credentials(current_user: CurrentUser = Depends(get_current_us
     Returns:
         SSH 연결 정보 및 임시 비밀번호
     """
+    # 1. 정적 자격증명 확인 (개발 환경 또는 정적 비밀번호 사용 시)
+    if settings.SSH_PASSWORD:
+        ssh_host = settings.SSH_HOST or "api.mouse84.com"
+        ssh_port = settings.SSH_PORT
+        ssh_username = settings.SSH_USERNAME or current_user.username
+        
+        # 만료 시간 (24시간)
+        expires_at = datetime.utcnow() + timedelta(hours=24)
+        
+        return {
+            "ssh_host": ssh_host,
+            "ssh_port": ssh_port,
+            "ssh_username": ssh_username,
+            "ssh_password": settings.SSH_PASSWORD,
+            "expires_at": expires_at.isoformat(),
+            "expires_in_seconds": 86400
+        }
+
+    # 2. 임시 비밀번호 생성 (기본 동작)
+    
+    # .env 설정을 우선 사용 (username이 지정된 경우 해당 계정으로 임시 비밀번호 생성)
+    target_username = settings.SSH_USERNAME or current_user.username
+    
     # 임시 비밀번호 생성 (1시간 유효)
     temp_password = temp_ssh_password_manager.generate_password(
-        username=current_user.username,
+        username=target_username,
         validity_hours=1
     )
     
-    # SSH 서버 정보 (외부 접속용)
-    ssh_host = os.getenv("SSH_HOST", "api.mouse84.com")  # 외부 DDNS
-    ssh_port = int(os.getenv("SSH_PORT", "54054"))  # 외부 포트 (내부 22로 포워딩)
+    # SSH 서버 정보 (설정값 우선)
+    ssh_host = settings.SSH_HOST or "api.mouse84.com"
+    ssh_port = settings.SSH_PORT or 54054
     
     # 만료 시간 계산
     expires_at = datetime.utcnow() + timedelta(hours=1)
@@ -37,7 +61,7 @@ async def get_ssh_credentials(current_user: CurrentUser = Depends(get_current_us
     return {
         "ssh_host": ssh_host,
         "ssh_port": ssh_port,
-        "ssh_username": current_user.username,
+        "ssh_username": target_username,
         "ssh_password": temp_password,
         "expires_at": expires_at.isoformat(),
         "expires_in_seconds": 3600

app/services/temp_ssh_password_service.py

@@ -1,14 +1,14 @@
 import secrets
 import hashlib
-from typing import Dict, Optional
+from typing import Dict, List, Optional
 from datetime import datetime, timedelta
 
 class TempSshPasswordManager:
     """임시 SSH 비밀번호 관리"""
     
     def __init__(self):
-        # 메모리 기반 저장소 (프로덕션에서는 Redis 사용 권장)
-        self._passwords: Dict[str, dict] = {}
+        # 메모리 기반 저장소 (username -> list of {hash, expires_at})
+        self._passwords: Dict[str, List[dict]] = {}
     
     def generate_password(self, username: str, validity_hours: int = 1) -> str:
         """
@@ -21,6 +21,9 @@ class TempSshPasswordManager:
         Returns:
             임시 비밀번호
         """
+        # 메모리 정리 (생성 시마다 만료된 것 정리)
+        self.cleanup_expired()
+
         # 안전한 랜덤 비밀번호 생성 (32자)
         temp_password = secrets.token_urlsafe(32)
         
@@ -30,49 +33,70 @@ class TempSshPasswordManager:
         # 만료 시간 계산
         expires_at = datetime.utcnow() + timedelta(hours=validity_hours)
         
-        # 저장
-        self._passwords[username] = {
+        # 새 토큰 정보
+        token_data = {
             "password_hash": password_hash,
             "expires_at": expires_at,
             "created_at": datetime.utcnow()
         }
+
+        # 해당 사용자에 토큰 추가 (리스트 초기화)
+        if username not in self._passwords:
+            self._passwords[username] = []
+        
+        self._passwords[username].append(token_data)
         
         return temp_password
     
     def verify_password(self, username: str, password: str) -> bool:
         """
-        비밀번호 검증
-        
-        Args:
-            username: 사용자명
-            password: 검증할 비밀번호
-            
-        Returns:
-            유효 여부
+        비밀번호 검증 (다중 토큰 지원)
         """
         if username not in self._passwords:
             return False
         
-        stored = self._passwords[username]
+        input_hash = hashlib.sha256(password.encode()).hexdigest()
+        now = datetime.utcnow()
         
-        # 만료 확인
-        if datetime.utcnow() > stored["expires_at"]:
-            del self._passwords[username]
-            return False
+        # 유효한 토큰 중 하나라도 일치하면 성공
+        # (리스트 복사본으로 순회하지 않고, 인덱스로 접근하거나 필터링)
+        valid_tokens = []
+        is_valid = False
         
-        # 비밀번호 확인
-        password_hash = hashlib.sha256(password.encode()).hexdigest()
-        return password_hash == stored["password_hash"]
+        for token in self._passwords[username]:
+            # 만료된 토큰은 제외 (Clean up on read)
+            if now > token["expires_at"]:
+                continue
+            
+            valid_tokens.append(token)
+            
+            if token["password_hash"] == input_hash:
+                is_valid = True
+                
+        # 리스트 업데이트 (만료된 것 제거됨)
+        self._passwords[username] = valid_tokens
+        
+        return is_valid
     
     def cleanup_expired(self):
         """만료된 비밀번호 정리"""
         now = datetime.utcnow()
-        expired = [
-            username for username, data in self._passwords.items()
-            if now > data["expires_at"]
-        ]
-        for username in expired:
-            del self._passwords[username]
+        users_to_check = list(self._passwords.keys())
+        
+        for username in users_to_check:
+            # 유효한 토큰만 필터링
+            self._passwords[username] = [
+                token for token in self._passwords[username] 
+                if now <= token["expires_at"]
+            ]
+            
+            # 토큰이 하나도 없으면 사용자 키 삭제
+            if not self._passwords[username]:
+                del self._passwords[username]
+                
+    def get_active_count(self) -> int:
+        """현재 활성화된 토큰 수 (디버깅용)"""
+        return sum(len(tokens) for tokens in self._passwords.values())
 
 # 싱글톤 인스턴스
 temp_ssh_password_manager = TempSshPasswordManager()

app/utils/security.py

@@ -1,12 +1,26 @@
-from passlib.context import CryptContext
+import bcrypt
 
-# bcrypt 컨텍스트
-pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+# bcrypt 컨텍스트 제거 (직접 사용)
 
 def hash_password(password: str) -> str:
     """비밀번호 해시화"""
-    return pwd_context.hash(password)
+    # bcrypt는 bytes를 처리하므로 인코딩 필요
+    pwd_bytes = password.encode('utf-8')
+    salt = bcrypt.gensalt()
+    hashed = bcrypt.hashpw(pwd_bytes, salt)
+    return hashed.decode('utf-8')
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:
     """비밀번호 검증"""
-    return pwd_context.verify(plain_password, hashed_password)
+    try:
+        if not plain_password or not hashed_password:
+            return False
+            
+        pwd_bytes = plain_password.encode('utf-8')
+        # DB에 저장된 해시는 str일 수 있으므로 인코딩
+        hashed_bytes = hashed_password.encode('utf-8')
+        
+        return bcrypt.checkpw(pwd_bytes, hashed_bytes)
+    except Exception:
+        # 형식 오류 등 발생 시 인증 실패 처리
+        return False

requirements.txt

@@ -10,9 +10,11 @@ alembic==1.13.1
 
 # Authentication & Security
 python-jose[cryptography]==3.3.0
-passlib[bcrypt]==1.7.4
+# passlib dependency removed
+bcrypt>=4.0.1
 python-dotenv==1.0.0
 pydantic-settings==2.1.0
+email-validator>=2.0.0
 
 # SSH & Networking
 paramiko==3.4.0