fix(bat-022): owner 재할당 credential 무효화 원자화 — leak 방지 (QC Major)

update_daemon_owner()에서 owner UPDATE + credential DELETE를 단일
conn.transaction() 블록으로 묶어 원자성 보장. 실패 시 두 연산 모두
롤백되므로 "새 owner + 이전 credential 잔류" leak 상태(dcy-011 OQ-3)
가 불가능해짐. 실행 순서도 DELETE-first(안전 순서)로 고정.

단위 테스트 3종 추가:
- DELETE-before-UPDATE 실행 순서 검증
- UPDATE 실패 시 rollback(commit 미호출) 검증
- 단일 transaction() 블록 안에서 두 연산 실행 검증

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VHtM5Ws1HKCKQ4JGi6GRb5

app/repositories/postgres.py

@@ -2385,20 +2385,37 @@ class PostgresPortfolioRepository:
     ) -> Daemon:
         """Change daemon owner and invalidate all existing BrokerCredentials (REQ-053 AC-4).
 
-        Owner change + credential invalidation are two separate statements (autocommit).
+        Owner change and credential invalidation are executed atomically in a single
-        Credential deletion failure is logged by caller — daemon update is treated as
+        transaction (BAT-022 QC Major fix).  If either statement fails the entire
-        authoritative.
+        transaction is rolled back, preventing the state where the owner has changed but
+        the previous owner's credentials remain (credential leak — dcy-011 OQ-3).
+
+        Execution order inside the transaction:
+          1. DELETE FROM broker_credentials WHERE daemon_id = %s   (credential first)
+          2. UPDATE daemons SET owner_user_id = %s ...             (owner second)
+        This order is also safe if the transaction machinery is somehow bypassed: if
+        only step 1 completes the owner is unchanged and credentials are gone (no leak);
+        if only step 2 completes the old credentials are still present but the caller
+        still owns them (no cross-owner leak).
         """
         self.ensure_schema()
-        with self._connect().cursor() as cur:
+        conn = self._connect()
+        # Use conn.transaction() (not `with conn:`) so the shared connection stays open
+        # after the block (psycopg3 closes the connection when used as a context manager).
+        with conn.transaction():
+            with conn.cursor() as cur:
                 cur.execute(
-                """
+                    "SELECT id FROM daemons WHERE id = %s",
-                SELECT id FROM daemons WHERE id = %s
-                """,
                     (daemon_id,),
                 )
                 if cur.fetchone() is None:
                     raise ValueError("데몬을 찾을 수 없습니다")
+                # Step 1: delete credentials first (safe-order)
+                cur.execute(
+                    "DELETE FROM broker_credentials WHERE daemon_id = %s",
+                    (daemon_id,),
+                )
+                # Step 2: update owner
                 cur.execute(
                     """
                     UPDATE daemons
@@ -2409,14 +2426,7 @@ class PostgresPortfolioRepository:
                     (owner_user_id, daemon_id),
                 )
                 row = cur.fetchone()
-        daemon = self._row_to_daemon(row)
+        return self._row_to_daemon(row)
-        # Invalidate broker credentials after owner change
-        with self._connect().cursor() as cur:
-            cur.execute(
-                "DELETE FROM broker_credentials WHERE daemon_id = %s",
-                (daemon_id,),
-            )
-        return daemon
 
     def list_service_tokens_with_null_daemon(self) -> list[AuthToken]:
         self.ensure_schema()

tests/test_bat022_req053_admin_daemon_crud.py

@@ -11,6 +11,7 @@ from fastapi.testclient import TestClient
 from app.core.config import Settings, get_settings
 from app.domain.models import BrokerCredential, Daemon, TokenPrincipal, User
 from app.main import create_app
+from app.repositories.postgres import PostgresPortfolioRepository
 
 _TEST_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
 
@@ -457,3 +458,196 @@ def test_admin_deactivate_nonexistent_daemon_returns_404() -> None:
         )
 
     assert resp.status_code == 404
+
+
+# ---------------------------------------------------------------------------
+# BAT-022 QC Major: update_daemon_owner 원자성 — 부분 실패 시 credential leak 방지
+# (dcy-011 OQ-3: 이전 owner의 KIS 인증정보 신규 owner에게 상속 불가)
+# ---------------------------------------------------------------------------
+
+def _make_mock_conn_for_owner_update(*, fail_on_update: bool = False):
+    """
+    psycopg Connection 모의 객체를 반환.
+    conn.transaction() 컨텍스트 매니저를 지원하고, 내부 cursor 실행을 추적한다.
+    fail_on_update=True 이면 UPDATE daemons 실행 시 예외를 발생시켜
+    트랜잭션 롤백을 유발한다.
+    """
+    from contextlib import contextmanager
+    from unittest.mock import MagicMock
+    import psycopg
+
+    executed_sqls: list[str] = []
+    rollback_called = False
+    commit_called = False
+
+    class TrackingCursor:
+        """execute 호출을 기록하고 SELECT/UPDATE/DELETE에 대한 결과를 제공."""
+        def __init__(self) -> None:
+            self._last_sql = ""
+            self._rows: list = []
+
+        def execute(self, sql: str, params=None) -> None:
+            sql_stripped = sql.strip()
+            executed_sqls.append(sql_stripped)
+            if sql_stripped.startswith("SELECT"):
+                # daemon exists
+                self._rows = [("daemon-001",)]
+            elif sql_stripped.startswith("DELETE"):
+                self._rows = []
+            elif sql_stripped.startswith("UPDATE"):
+                if fail_on_update:
+                    raise psycopg.DatabaseError("simulated UPDATE failure")
+                self._rows = [(
+                    "daemon-001",
+                    "user-b",
+                    "trex-orb",
+                    "ORB",
+                    True,
+                    datetime(2026, 6, 28, tzinfo=timezone.utc),
+                    datetime(2026, 6, 28, tzinfo=timezone.utc),
+                )]
+
+        def fetchone(self):
+            if self._rows:
+                return self._rows[0]
+            return None
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            pass
+
+    class FakeTransaction:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc_val, exc_tb):
+            nonlocal rollback_called, commit_called
+            if exc_type is not None:
+                rollback_called = True
+                return False  # re-raise
+            commit_called = True
+            return False
+
+    class FakeConn:
+        def transaction(self):
+            return FakeTransaction()
+
+        def cursor(self):
+            return TrackingCursor()
+
+    return FakeConn(), executed_sqls, lambda: rollback_called, lambda: commit_called
+
+
+def test_update_daemon_owner_executes_delete_before_update() -> None:
+    """BAT-022 QC Major / dcy-011 OQ-3: credential DELETE가 owner UPDATE보다 먼저 실행됨을 확인.
+
+    안전 순서 보장: DELETE가 먼저이므로 UPDATE 실패 시 owner 미변경+credential 없음(leak 없음).
+    """
+    from unittest.mock import patch as _patch
+
+    fake_conn, executed_sqls, is_rollback, is_commit = _make_mock_conn_for_owner_update()
+
+    repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
+
+    with _patch.object(repo, "_connect", return_value=fake_conn), \
+         _patch.object(repo, "ensure_schema"):
+        result = repo.update_daemon_owner("daemon-001", owner_user_id="user-b")
+
+    # DELETE must come before UPDATE in execution order
+    delete_idx = next(i for i, s in enumerate(executed_sqls) if s.startswith("DELETE"))
+    update_idx = next(i for i, s in enumerate(executed_sqls) if s.startswith("UPDATE"))
+    assert delete_idx < update_idx, (
+        f"DELETE must execute before UPDATE to prevent credential leak. "
+        f"Actual order: DELETE={delete_idx}, UPDATE={update_idx}"
+    )
+    # Owner change committed
+    assert is_commit()
+    assert result.owner_user_id == "user-b"
+
+
+def test_update_daemon_owner_rollback_on_update_failure_prevents_leak() -> None:
+    """BAT-022 QC Major / dcy-011 OQ-3: UPDATE daemons 실패 시 트랜잭션 롤백(rollback 경로 확인).
+
+    부분 실패 시 "새 owner + 이전 credential 잔류" 상태(credential leak)가 발생하지 않아야 함.
+    트랜잭션 블록 안에서 UPDATE가 실패하면 예외가 전파되고 commit은 호출되지 않는다.
+    """
+    import psycopg
+    from unittest.mock import patch as _patch
+
+    fake_conn, executed_sqls, is_rollback, is_commit = _make_mock_conn_for_owner_update(fail_on_update=True)
+
+    repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
+
+    with _patch.object(repo, "_connect", return_value=fake_conn), \
+         _patch.object(repo, "ensure_schema"):
+        with pytest.raises(psycopg.DatabaseError):
+            repo.update_daemon_owner("daemon-001", owner_user_id="user-b")
+
+    # DELETE was attempted before UPDATE
+    assert any(s.startswith("DELETE") for s in executed_sqls), \
+        "DELETE must have been attempted before UPDATE"
+    # UPDATE failure triggered rollback (not commit)
+    assert is_rollback(), "Transaction must roll back when UPDATE fails"
+    assert not is_commit(), "Transaction must NOT commit on UPDATE failure"
+
+
+def test_update_daemon_owner_both_ops_in_single_transaction() -> None:
+    """BAT-022 QC Major: DELETE + UPDATE가 단일 conn.transaction() 블록 안에서 실행됨을 확인.
+
+    두 연산이 별도 커넥션이 아닌 같은 트랜잭션 안에 묶여 있어야 원자성이 보장된다.
+    """
+    from unittest.mock import patch as _patch, call as _call
+
+    transaction_enter_count = [0]
+    executed_sqls: list[str] = []
+
+    class CountingTransaction:
+        def __enter__(self):
+            transaction_enter_count[0] += 1
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+    class TrackingCursor:
+        def execute(self, sql: str, params=None) -> None:
+            executed_sqls.append(sql.strip())
+
+        def fetchone(self):
+            sql = executed_sqls[-1] if executed_sqls else ""
+            if sql.startswith("SELECT"):
+                return ("daemon-001",)
+            if sql.startswith("UPDATE"):
+                return (
+                    "daemon-001", "user-b", "trex-orb", "ORB", True,
+                    datetime(2026, 6, 28, tzinfo=timezone.utc),
+                    datetime(2026, 6, 28, tzinfo=timezone.utc),
+                )
+            return None
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            pass
+
+    class FakeConn:
+        def transaction(self):
+            return CountingTransaction()
+
+        def cursor(self):
+            return TrackingCursor()
+
+    repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
+
+    with _patch.object(repo, "_connect", return_value=FakeConn()), \
+         _patch.object(repo, "ensure_schema"):
+        repo.update_daemon_owner("daemon-001", owner_user_id="user-b")
+
+    # Exactly one transaction block entered (not two separate autocommit calls)
+    assert transaction_enter_count[0] == 1, (
+        f"Expected 1 transaction block, got {transaction_enter_count[0]}. "
+        "DELETE and UPDATE must be inside a single transaction."
+    )