feat(bat-020): daemon-scoped 브로커 인증정보 (REQ-046~051)

- REQ-046: BrokerCredential.daemon_id 필드 추가, repository 시그니처 갱신
- REQ-047: alembic revision Step1(daemon_id NULL 컬럼 추가), Step2(PK 전환, dev/test DB 한정)
- REQ-048: backfill-daemon-id CLI subcommand (--dry-run/--force/--broker 지원)
- REQ-049: GET /v1/user_info daemon-scoped 단일 broker 전달 (brokers→broker, daemon_id 없으면 403)
- REQ-050: credential set/list/remove --daemon-id 옵션 추가 (daemon_slug 해석)
- REQ-051: 보안 격리 - service-token 전용, web 토큰 차단, 로그 app_secret 비노출, graceful degrade

비가역 결재 게이트: REQ-047 Step2(PK 전환)는 사장님 결재 후 실DB 적용 예정(이 batch는 revision 코드만)

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

tests/test_bat020_broker_credential_domain.py

@@ -0,0 +1,178 @@
+"""BAT-020: REQ-046 BrokerCredential daemon_id 필드 단위 테스트."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+
+from app.domain.models import BrokerCredential
+
+
+# ---------------------------------------------------------------------------
+# REQ-046 AC-1: daemon_id=None 기본값
+# ---------------------------------------------------------------------------
+
+def test_broker_credential_daemon_id_defaults_to_none() -> None:
+    """REQ-046 AC-1: daemon_id 미지정 시 None이 기본값."""
+    cred = BrokerCredential(
+        user_id="user-1",
+        broker="kis",
+        account_ref="12345678",
+        encrypted_payload="encrypted",
+    )
+    assert hasattr(cred, "daemon_id")
+    assert cred.daemon_id is None
+
+
+# ---------------------------------------------------------------------------
+# REQ-046 AC-2: daemon_id 값 지정
+# ---------------------------------------------------------------------------
+
+def test_broker_credential_daemon_id_set() -> None:
+    """REQ-046 AC-2: daemon_id="daemon-abc" 지정 시 해당 값으로 세팅."""
+    cred = BrokerCredential(
+        user_id="user-1",
+        broker="kis",
+        account_ref="12345678",
+        encrypted_payload="encrypted",
+        daemon_id="daemon-abc",
+    )
+    assert cred.daemon_id == "daemon-abc"
+
+
+# ---------------------------------------------------------------------------
+# REQ-046 AC-3: _row_to_broker_credential 7컬럼(daemon_id 포함) 처리
+# ---------------------------------------------------------------------------
+
+def test_row_to_broker_credential_with_daemon_id() -> None:
+    """REQ-046 AC-3: 7컬럼 행(daemon_id 포함) 변환 시 daemon_id 세팅."""
+    from app.repositories.postgres import PostgresPortfolioRepository
+
+    now = datetime.now(timezone.utc)
+    row = ("user-1", "kis", "12345678", "encrypted", 2, now, "daemon-xyz")
+    cred = PostgresPortfolioRepository._row_to_broker_credential(row)
+    assert cred.user_id == "user-1"
+    assert cred.broker == "kis"
+    assert cred.account_ref == "12345678"
+    assert cred.encrypted_payload == "encrypted"
+    assert cred.credential_version == 2
+    assert cred.last_rotated_at == now
+    assert cred.daemon_id == "daemon-xyz"
+
+
+# ---------------------------------------------------------------------------
+# REQ-046 AC-4: _row_to_broker_credential 6컬럼(daemon_id 없음) 하위호환
+# ---------------------------------------------------------------------------
+
+def test_row_to_broker_credential_without_daemon_id() -> None:
+    """REQ-046 AC-4: 6컬럼 행(daemon_id 없음) 변환 시 daemon_id=None (하위호환)."""
+    from app.repositories.postgres import PostgresPortfolioRepository
+
+    now = datetime.now(timezone.utc)
+    row = ("user-1", "kis", "12345678", "encrypted", 1, now)
+    cred = PostgresPortfolioRepository._row_to_broker_credential(row)
+    assert cred.daemon_id is None
+
+
+# ---------------------------------------------------------------------------
+# REQ-046 AC-5/6: list_broker_credentials daemon_id 필터 (mock 기반)
+# ---------------------------------------------------------------------------
+
+class _FakeCursor:
+    """psycopg Cursor 최소 mock."""
+    def __init__(self, rows):
+        self._rows = rows
+        self.last_query = None
+        self.last_params = None
+
+    def execute(self, query, params=None):
+        self.last_query = query
+        self.last_params = params
+
+    def fetchall(self):
+        return self._rows
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        pass
+
+
+class _FakeConn:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def cursor(self):
+        return _FakeCursor(self._rows)
+
+
+def _make_rows_for_user(user_id: str):
+    """daemon_id "d1", "d2" 각각 1건씩."""
+    now = datetime.now(timezone.utc)
+    return [
+        (user_id, "kis", "ref-d1", "enc-d1", 1, now, "d1"),
+        (user_id, "kis2", "ref-d2", "enc-d2", 1, now, "d2"),
+    ]
+
+
+def test_list_broker_credentials_daemon_id_filter() -> None:
+    """REQ-046 AC-5: daemon_id 필터 시 해당 daemon_id 행만 반환."""
+    from unittest.mock import MagicMock, patch
+    from app.repositories.postgres import PostgresPortfolioRepository, PortfolioDBConfig
+
+    rows = _make_rows_for_user("user-1")
+    # daemon_id="d1" 필터 시 d1 행만 반환되는 mock
+    d1_rows = [rows[0]]
+
+    config = PortfolioDBConfig(
+        host="localhost", port=5432, dbname="test",
+        user="test", password="test", sslmode="disable",
+    )
+    repo = PostgresPortfolioRepository(config)
+
+    mock_cursor = MagicMock()
+    mock_cursor.__enter__ = lambda s: mock_cursor
+    mock_cursor.__exit__ = MagicMock(return_value=False)
+    mock_cursor.fetchall.return_value = d1_rows
+
+    mock_conn = MagicMock()
+    mock_conn.cursor.return_value = mock_cursor
+    repo._connection = mock_conn
+
+    # ensure_schema 무력화
+    with patch.object(repo, "ensure_schema"):
+        results = repo.list_broker_credentials("user-1", daemon_id="d1")
+
+    assert len(results) == 1
+    assert results[0].daemon_id == "d1"
+    assert results[0].broker == "kis"
+
+
+def test_list_broker_credentials_no_daemon_id_filter_returns_all() -> None:
+    """REQ-046 AC-6: daemon_id 미지정 시 전체 반환 (하위호환)."""
+    from unittest.mock import MagicMock, patch
+    from app.repositories.postgres import PostgresPortfolioRepository, PortfolioDBConfig
+
+    rows = _make_rows_for_user("user-1")
+
+    config = PortfolioDBConfig(
+        host="localhost", port=5432, dbname="test",
+        user="test", password="test", sslmode="disable",
+    )
+    repo = PostgresPortfolioRepository(config)
+
+    mock_cursor = MagicMock()
+    mock_cursor.__enter__ = lambda s: mock_cursor
+    mock_cursor.__exit__ = MagicMock(return_value=False)
+    mock_cursor.fetchall.return_value = rows
+
+    mock_conn = MagicMock()
+    mock_conn.cursor.return_value = mock_cursor
+    repo._connection = mock_conn
+
+    with patch.object(repo, "ensure_schema"):
+        results = repo.list_broker_credentials("user-1")
+
+    assert len(results) == 2

tests/test_bat020_credential_cli.py

@@ -0,0 +1,460 @@
+"""BAT-020: REQ-050, REQ-048 — credential CLI --daemon-id + backfill 테스트."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from typing import Any
+
+import pytest
+from typer.testing import CliRunner
+
+from app.cli.main import create_app
+from app.core.config import Settings
+from app.core.encryption import encrypt_payload
+from app.core.security import hash_token, token_prefix
+from app.domain.models import AuthToken, BrokerCredential, Daemon, User
+
+
+_TEST_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+
+
+# ---------------------------------------------------------------------------
+# Minimal FakeRepo for credential CLI tests
+# ---------------------------------------------------------------------------
+
+class CredFakeRepo:
+    """credential CLI 테스트용 최소 FakeRepo."""
+
+    def __init__(self) -> None:
+        self._users: dict[str, User] = {}
+        self._daemons: dict[str, Daemon] = {}
+        self._broker_credentials: dict[tuple, BrokerCredential] = {}
+        self._seq = 1
+
+    def close(self) -> None:
+        pass
+
+    def ping(self) -> None:
+        pass
+
+    def ensure_schema(self) -> None:
+        pass
+
+    def get_user(self, user_id: str) -> User:
+        u = self._users.get(user_id)
+        if u is None:
+            raise ValueError("사용자를 찾을 수 없습니다")
+        return u
+
+    def get_user_by_username(self, username: str) -> User | None:
+        return next((u for u in self._users.values() if u.username == username), None)
+
+    def create_user(self, username: str, **kwargs) -> User:
+        user = User(
+            id=f"user-{self._seq}",
+            username=username,
+            email=kwargs.get("email"),
+            permission=kwargs.get("permission", "read"),
+            is_active=kwargs.get("is_active", True),
+        )
+        self._seq += 1
+        self._users[user.id] = user
+        return user
+
+    def create_daemon(self, owner_user_id: str, *, daemon_slug: str, display_name: str | None = None) -> Daemon:
+        d = Daemon(
+            id=f"daemon-{self._seq}",
+            owner_user_id=owner_user_id,
+            daemon_slug=daemon_slug,
+            display_name=display_name,
+            is_active=True,
+            created_at=datetime.now(timezone.utc),
+        )
+        self._seq += 1
+        self._daemons[d.id] = d
+        return d
+
+    def get_daemon_by_slug(self, owner_user_id: str, daemon_slug: str) -> Daemon | None:
+        return next(
+            (d for d in self._daemons.values() if d.owner_user_id == owner_user_id and d.daemon_slug == daemon_slug),
+            None,
+        )
+
+    def get_daemon_by_id(self, daemon_id: str) -> Daemon | None:
+        return self._daemons.get(daemon_id)
+
+    def upsert_broker_credential(
+        self,
+        user_id: str,
+        broker: str,
+        *,
+        account_ref: str,
+        encrypted_payload: str,
+        daemon_id: str | None = None,
+    ) -> BrokerCredential:
+        self.get_user(user_id)
+        key = (user_id, broker)
+        existing = self._broker_credentials.get(key)
+        cred = BrokerCredential(
+            user_id=user_id,
+            broker=broker,
+            account_ref=account_ref,
+            encrypted_payload=encrypted_payload,
+            credential_version=(existing.credential_version + 1) if existing else 1,
+            last_rotated_at=datetime.now(timezone.utc),
+            daemon_id=daemon_id,
+        )
+        self._broker_credentials[key] = cred
+        return cred
+
+    def list_broker_credentials(
+        self,
+        user_id: str,
+        daemon_id: str | None = None,
+    ) -> list[BrokerCredential]:
+        creds = [c for (uid, _), c in self._broker_credentials.items() if uid == user_id]
+        if daemon_id is not None:
+            creds = [c for c in creds if c.daemon_id == daemon_id]
+        return sorted(creds, key=lambda c: c.broker)
+
+    def update_broker_credential_daemon_id(self, user_id: str, broker: str, daemon_id: str) -> bool:
+        key = (user_id, broker)
+        cred = self._broker_credentials.get(key)
+        if cred is None or cred.daemon_id is not None:
+            return False
+        self._broker_credentials[key] = BrokerCredential(
+            user_id=cred.user_id, broker=cred.broker, account_ref=cred.account_ref,
+            encrypted_payload=cred.encrypted_payload, credential_version=cred.credential_version,
+            last_rotated_at=cred.last_rotated_at, daemon_id=daemon_id,
+        )
+        return True
+
+    def delete_broker_credential(self, user_id: str, broker: str, daemon_id: str | None = None) -> bool:
+        if daemon_id is not None:
+            for key, cred in list(self._broker_credentials.items()):
+                if cred.daemon_id == daemon_id and cred.broker == broker:
+                    del self._broker_credentials[key]
+                    return True
+            return False
+        key = (user_id, broker)
+        if key not in self._broker_credentials:
+            return False
+        del self._broker_credentials[key]
+        return True
+
+
+def _settings_factory() -> Settings:
+    return Settings(
+        allowed_origins_raw="http://localhost:3000",
+        credential_encryption_key=_TEST_ENC_KEY,
+    )
+
+
+def _build_runner(repo: CredFakeRepo) -> Any:
+    app = create_app(settings_factory=_settings_factory, repository_factory=lambda settings: repo)
+    runner = CliRunner()
+    runner.app = app
+    return runner
+
+
+def _make_encrypted(app_key: str = "k", app_secret: str = "s") -> str:
+    return encrypt_payload(_TEST_ENC_KEY, json.dumps({"app_key": app_key, "app_secret": app_secret}))
+
+
+# ---------------------------------------------------------------------------
+# REQ-050 AC-1: credential set --daemon-id 성공
+# ---------------------------------------------------------------------------
+
+def test_credential_set_with_daemon_id() -> None:
+    """REQ-050 AC-1: set --daemon-id → broker_credentials에 daemon_id 포함 upsert."""
+    repo = CredFakeRepo()
+    user = repo.create_user("alice")
+    daemon = repo.create_daemon(user.id, daemon_slug="orb-bot")
+    runner = _build_runner(repo)
+
+    result = runner.invoke(runner.app, [
+        "--format", "json", "credential", "set",
+        "--user", "alice",
+        "--daemon-id", "orb-bot",
+        "--broker", "kis",
+        "--account-ref", "acc-123",
+        "--app-key", "mykey",
+        "--app-secret", "mysecret",
+        "--force",
+    ])
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.stdout)
+    assert payload["broker"] == "kis"
+    assert payload["daemon_id"] == daemon.id
+    assert payload["account_ref"] == "acc-123"
+
+
+# ---------------------------------------------------------------------------
+# REQ-050 AC-2: credential list --daemon-id 필터
+# ---------------------------------------------------------------------------
+
+def test_credential_list_daemon_id_filter() -> None:
+    """REQ-050 AC-2: list --daemon-id → 해당 daemon 인증정보만 출력."""
+    repo = CredFakeRepo()
+    user = repo.create_user("alice")
+    d1 = repo.create_daemon(user.id, daemon_slug="bot-d1")
+    d2 = repo.create_daemon(user.id, daemon_slug="bot-d2")
+
+    enc = _make_encrypted()
+    repo.upsert_broker_credential(user.id, "kis", account_ref="ref-d1", encrypted_payload=enc, daemon_id=d1.id)
+    repo.upsert_broker_credential(user.id, "kis2", account_ref="ref-d2", encrypted_payload=enc, daemon_id=d2.id)
+
+    runner = _build_runner(repo)
+    result = runner.invoke(runner.app, [
+        "--format", "json", "credential", "list",
+        "--user", "alice",
+        "--daemon-id", "bot-d1",
+    ])
+
+    assert result.exit_code == 0, result.output
+    items = json.loads(result.stdout)
+    assert len(items) == 1
+    assert items[0]["daemon_id"] == d1.id
+
+
+# ---------------------------------------------------------------------------
+# REQ-050 AC-3: credential list 전체 (daemon-id 미지정)
+# ---------------------------------------------------------------------------
+
+def test_credential_list_no_daemon_id_returns_all() -> None:
+    """REQ-050 AC-3: list --daemon-id 미지정 → 전체 인증정보."""
+    repo = CredFakeRepo()
+    user = repo.create_user("alice")
+    d1 = repo.create_daemon(user.id, daemon_slug="bot-d1")
+    d2 = repo.create_daemon(user.id, daemon_slug="bot-d2")
+
+    enc = _make_encrypted()
+    repo.upsert_broker_credential(user.id, "kis", account_ref="ref-d1", encrypted_payload=enc, daemon_id=d1.id)
+    repo.upsert_broker_credential(user.id, "kis2", account_ref="ref-d2", encrypted_payload=enc, daemon_id=d2.id)
+
+    runner = _build_runner(repo)
+    result = runner.invoke(runner.app, [
+        "--format", "json", "credential", "list",
+        "--user", "alice",
+    ])
+
+    assert result.exit_code == 0, result.output
+    items = json.loads(result.stdout)
+    assert len(items) == 2
+
+
+# ---------------------------------------------------------------------------
+# REQ-050 AC-4: credential remove --daemon-id
+# ---------------------------------------------------------------------------
+
+def test_credential_remove_with_daemon_id() -> None:
+    """REQ-050 AC-4: remove --daemon-id → (daemon_id, broker) 기반 삭제."""
+    repo = CredFakeRepo()
+    user = repo.create_user("alice")
+    d1 = repo.create_daemon(user.id, daemon_slug="orb-bot")
+
+    enc = _make_encrypted()
+    repo.upsert_broker_credential(user.id, "kis", account_ref="ref", encrypted_payload=enc, daemon_id=d1.id)
+
+    runner = _build_runner(repo)
+    result = runner.invoke(runner.app, [
+        "--format", "json", "credential", "remove",
+        "--user", "alice",
+        "--daemon-id", "orb-bot",
+        "--broker", "kis",
+        "--force",
+    ])
+
+    assert result.exit_code == 0, result.output
+    remaining = repo.list_broker_credentials(user.id)
+    assert len(remaining) == 0
+
+
+# ---------------------------------------------------------------------------
+# REQ-050 AC-5: 존재하지 않는 daemon_slug → exit 1
+# ---------------------------------------------------------------------------
+
+def test_credential_set_nonexistent_daemon_exits_1() -> None:
+    """REQ-050 AC-5: 존재하지 않는 daemon_slug → 오류 + exit 1."""
+    repo = CredFakeRepo()
+    repo.create_user("alice")
+
+    runner = _build_runner(repo)
+    result = runner.invoke(runner.app, [
+        "--format", "json", "credential", "set",
+        "--user", "alice",
+        "--daemon-id", "nonexistent-bot",
+        "--broker", "kis",
+        "--account-ref", "acc",
+        "--app-key", "k",
+        "--app-secret", "s",
+        "--force",
+    ])
+
+    assert result.exit_code != 0
+
+
+# ---------------------------------------------------------------------------
+# REQ-050 AC-7: CLI 출력에 daemon_id 포함
+# ---------------------------------------------------------------------------
+
+def test_credential_set_output_includes_daemon_id() -> None:
+    """REQ-050 AC-7: set 성공 출력에 daemon_id 포함."""
+    repo = CredFakeRepo()
+    user = repo.create_user("alice")
+    daemon = repo.create_daemon(user.id, daemon_slug="tsl-bot")
+    runner = _build_runner(repo)
+
+    result = runner.invoke(runner.app, [
+        "--format", "json", "credential", "set",
+        "--user", "alice",
+        "--daemon-id", "tsl-bot",
+        "--broker", "kis",
+        "--account-ref", "acc-999",
+        "--app-key", "k2",
+        "--app-secret", "s2",
+        "--force",
+    ])
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.stdout)
+    assert "daemon_id" in payload
+    assert "broker" in payload
+    assert "account_ref" in payload
+    assert "credential_version" in payload
+    assert "last_rotated_at" in payload
+
+
+# ---------------------------------------------------------------------------
+# REQ-048 AC-1/2: backfill-daemon-id 정상 백필
+# ---------------------------------------------------------------------------
+
+def test_backfill_daemon_id_updates_null_rows() -> None:
+    """REQ-048 AC-1/2: daemon_id IS NULL 인증정보 → 백필 후 daemon_id 세팅."""
+    repo = CredFakeRepo()
+    user = repo.create_user("alice")
+    daemon = repo.create_daemon(user.id, daemon_slug="orb-bot")
+
+    enc = _make_encrypted()
+    # daemon_id=None으로 등록 (기존 user-scoped 방식)
+    repo.upsert_broker_credential(user.id, "kis", account_ref="ref", encrypted_payload=enc, daemon_id=None)
+
+    runner = _build_runner(repo)
+    result = runner.invoke(runner.app, [
+        "--format", "json", "credential", "backfill-daemon-id",
+        "--user", "alice",
+        "--daemon-id", "orb-bot",
+        "--force",
+    ])
+
+    assert result.exit_code == 0, result.output
+    # 백필 후 daemon_id 세팅 확인
+    updated = repo.list_broker_credentials(user.id)
+    assert len(updated) == 1
+    assert updated[0].daemon_id == daemon.id
+
+
+# ---------------------------------------------------------------------------
+# REQ-048 AC-3: --dry-run 시 DB 변경 없음
+# ---------------------------------------------------------------------------
+
+def test_backfill_daemon_id_dry_run_no_change() -> None:
+    """REQ-048 AC-3: --dry-run → 실제 UPDATE 없음."""
+    repo = CredFakeRepo()
+    user = repo.create_user("alice")
+    daemon = repo.create_daemon(user.id, daemon_slug="orb-bot")
+
+    enc = _make_encrypted()
+    repo.upsert_broker_credential(user.id, "kis", account_ref="ref", encrypted_payload=enc, daemon_id=None)
+
+    runner = _build_runner(repo)
+    result = runner.invoke(runner.app, [
+        "--format", "json", "credential", "backfill-daemon-id",
+        "--user", "alice",
+        "--daemon-id", "orb-bot",
+        "--dry-run",
+    ])
+
+    assert result.exit_code == 0, result.output
+    # DB 변경 없음 (dry-run)
+    creds = repo.list_broker_credentials(user.id)
+    assert creds[0].daemon_id is None
+
+
+# ---------------------------------------------------------------------------
+# REQ-048 AC-5: daemon_id IS NULL 없으면 "이미 완료됨"
+# ---------------------------------------------------------------------------
+
+def test_backfill_daemon_id_already_complete() -> None:
+    """REQ-048 AC-5: daemon_id IS NULL 없으면 이미 완료됨 + exit 0."""
+    repo = CredFakeRepo()
+    user = repo.create_user("alice")
+    daemon = repo.create_daemon(user.id, daemon_slug="orb-bot")
+
+    enc = _make_encrypted()
+    # 이미 daemon_id 있음
+    repo.upsert_broker_credential(user.id, "kis", account_ref="ref", encrypted_payload=enc, daemon_id=daemon.id)
+
+    runner = _build_runner(repo)
+    result = runner.invoke(runner.app, [
+        "--format", "json", "credential", "backfill-daemon-id",
+        "--user", "alice",
+        "--daemon-id", "orb-bot",
+        "--force",
+    ])
+
+    assert result.exit_code == 0, result.output
+    assert "already_complete" in result.stdout or "이미 완료됨" in result.stdout
+
+
+# ---------------------------------------------------------------------------
+# REQ-048 AC-6: 존재하지 않는 user → exit 1
+# ---------------------------------------------------------------------------
+
+def test_backfill_daemon_id_nonexistent_user_exits_1() -> None:
+    """REQ-048 AC-6: 존재하지 않는 user --user → exit 1."""
+    repo = CredFakeRepo()
+    repo.create_daemon.__doc__  # dummy
+
+    runner = _build_runner(repo)
+    result = runner.invoke(runner.app, [
+        "--format", "json", "credential", "backfill-daemon-id",
+        "--user", "nobody",
+        "--daemon-id", "orb-bot",
+        "--force",
+    ])
+
+    assert result.exit_code != 0
+
+
+# ---------------------------------------------------------------------------
+# REQ-048 AC-4: 백필 전후 encrypted_payload 동등 (데이터 손실 없음)
+# ---------------------------------------------------------------------------
+
+def test_backfill_preserves_encrypted_payload() -> None:
+    """REQ-048 AC-4: 백필 전후 encrypted_payload, account_ref 동등."""
+    repo = CredFakeRepo()
+    user = repo.create_user("alice")
+    daemon = repo.create_daemon(user.id, daemon_slug="orb-bot")
+
+    enc = _make_encrypted("orig_key", "orig_secret")
+    repo.upsert_broker_credential(
+        user.id, "kis", account_ref="original-ref", encrypted_payload=enc, daemon_id=None
+    )
+
+    original_encrypted = enc
+    original_account_ref = "original-ref"
+
+    runner = _build_runner(repo)
+    runner.invoke(runner.app, [
+        "--format", "json", "credential", "backfill-daemon-id",
+        "--user", "alice",
+        "--daemon-id", "orb-bot",
+        "--force",
+    ])
+
+    creds = repo.list_broker_credentials(user.id)
+    assert creds[0].encrypted_payload == original_encrypted
+    assert creds[0].account_ref == original_account_ref

tests/test_bat020_user_info_daemon_scoped.py

@@ -0,0 +1,436 @@
+"""BAT-020: REQ-049, REQ-051 — GET /v1/user_info daemon-scoped + 보안 격리 테스트."""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.core.auth import AuthContext
+from app.core.config import Settings, get_settings
+from app.core.encryption import encrypt_payload
+from app.domain.models import BrokerCredential, TokenPrincipal, User
+from app.main import create_app
+
+_TEST_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+
+
+def _make_settings(**kwargs) -> Settings:
+    return Settings(
+        allowed_origins_raw="http://localhost:3000",
+        db_enabled=True,
+        credential_encryption_key=_TEST_ENC_KEY,
+        **kwargs,
+    )
+
+
+def _make_client(settings: Settings, mock_repo) -> TestClient:
+    get_settings.cache_clear()
+    app = create_app(settings_factory=lambda: settings)
+    app.dependency_overrides[get_settings] = lambda: settings
+    return TestClient(app)
+
+
+def _svc_token(token: str = "svc-token") -> dict:
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _make_cred(user_id: str, daemon_id: str, broker: str = "kis", account_ref: str = "ref") -> BrokerCredential:
+    plain = json.dumps({"app_key": "k", "app_secret": "s"})
+    encrypted = encrypt_payload(_TEST_ENC_KEY, plain)
+    return BrokerCredential(
+        user_id=user_id, broker=broker, account_ref=account_ref,
+        encrypted_payload=encrypted, credential_version=1,
+        daemon_id=daemon_id,
+    )
+
+
+# ---------------------------------------------------------------------------
+# REQ-049 AC-1: daemon_id 있음 + broker 있음 → 200 + 단일 broker
+# ---------------------------------------------------------------------------
+
+def test_user_info_daemon_scoped_broker_returned() -> None:
+    """REQ-049 AC-1: daemon_id 바인딩 token + 해당 daemon broker 있음 → 단일 BrokerItem."""
+    settings = _make_settings()
+    principal = TokenPrincipal(
+        token_id="t1", user_id="u1", permission="read",
+        token_type="service", runtime_id="rt-1", daemon_id="d1",
+    )
+    user = User(id="u1", username="alice", permission="read", is_active=True)
+    cred = _make_cred("u1", "d1", account_ref="acc-d1")
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+    mock_repo.get_user.return_value = user
+    mock_repo.list_broker_credentials.return_value = [cred]
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        resp = client.get("/v1/user_info", headers=_svc_token())
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["broker"] is not None
+    assert body["broker"]["account_ref"] == "acc-d1"
+    assert body["broker"]["broker"] == "kis"
+    assert body["broker"]["credential_payload"] == {"app_key": "k", "app_secret": "s"}
+    # REQ-049 AC-6: brokers 키 없음
+    assert "brokers" not in body
+
+
+# ---------------------------------------------------------------------------
+# REQ-049 AC-2: daemon_id 있음 + broker 없음 → 200 + broker=null
+# ---------------------------------------------------------------------------
+
+def test_user_info_daemon_no_broker_returns_null() -> None:
+    """REQ-049 AC-2: daemon_id 있음, 해당 daemon 인증정보 없음 → broker=null."""
+    settings = _make_settings()
+    principal = TokenPrincipal(
+        token_id="t1", user_id="u1", permission="read",
+        token_type="service", daemon_id="d1",
+    )
+    user = User(id="u1", username="alice", permission="read", is_active=True)
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+    mock_repo.get_user.return_value = user
+    mock_repo.list_broker_credentials.return_value = []  # 인증정보 없음
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        resp = client.get("/v1/user_info", headers=_svc_token())
+
+    assert resp.status_code == 200
+    assert resp.json()["broker"] is None
+
+
+# ---------------------------------------------------------------------------
+# REQ-049 AC-3, REQ-051 AC-3: daemon_id 없는 service token → 403
+# ---------------------------------------------------------------------------
+
+def test_user_info_no_daemon_id_returns_403() -> None:
+    """REQ-049 AC-3, REQ-051 AC-3: daemon_id=None service token → 403."""
+    settings = _make_settings()
+    principal = TokenPrincipal(
+        token_id="t1", user_id="u1", permission="read",
+        token_type="service", daemon_id=None,
+    )
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        resp = client.get("/v1/user_info", headers=_svc_token())
+
+    assert resp.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# REQ-049 AC-4/5: 같은 user 2데몬 격리
+# ---------------------------------------------------------------------------
+
+def test_user_info_daemon_isolation_two_daemons() -> None:
+    """REQ-049 AC-4/5: daemon A, B가 각각 다른 account_ref를 반환."""
+    settings = _make_settings()
+    user = User(id="u1", username="alice", permission="read", is_active=True)
+    cred_a = _make_cred("u1", "d-a", account_ref="acc-A")
+    cred_b = _make_cred("u1", "d-b", account_ref="acc-B")
+
+    principal_a = TokenPrincipal(
+        token_id="ta", user_id="u1", permission="read",
+        token_type="service", daemon_id="d-a",
+    )
+    principal_b = TokenPrincipal(
+        token_id="tb", user_id="u1", permission="read",
+        token_type="service", daemon_id="d-b",
+    )
+
+    # daemon A 호출
+    mock_repo_a = MagicMock()
+    mock_repo_a.get_token_principal.return_value = principal_a
+    mock_repo_a.get_user.return_value = user
+    mock_repo_a.list_broker_credentials.return_value = [cred_a]
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo_a), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo_a):
+        client = _make_client(settings, mock_repo_a)
+        resp_a = client.get("/v1/user_info", headers=_svc_token())
+
+    assert resp_a.status_code == 200
+    assert resp_a.json()["broker"]["account_ref"] == "acc-A"
+
+    # daemon B 호출
+    mock_repo_b = MagicMock()
+    mock_repo_b.get_token_principal.return_value = principal_b
+    mock_repo_b.get_user.return_value = user
+    mock_repo_b.list_broker_credentials.return_value = [cred_b]
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo_b), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo_b):
+        client_b = _make_client(settings, mock_repo_b)
+        resp_b = client_b.get("/v1/user_info", headers=_svc_token())
+
+    assert resp_b.status_code == 200
+    assert resp_b.json()["broker"]["account_ref"] == "acc-B"
+    # 상호 반환 없음
+    assert resp_a.json()["broker"]["account_ref"] != resp_b.json()["broker"]["account_ref"]
+
+
+# ---------------------------------------------------------------------------
+# REQ-049 AC-6: 응답에 broker 단일 키 존재, brokers 리스트 없음
+# ---------------------------------------------------------------------------
+
+def test_user_info_response_has_broker_not_brokers() -> None:
+    """REQ-049 AC-6: 응답 JSON에 broker 단일 키만 있고 brokers 키 없음."""
+    settings = _make_settings()
+    principal = TokenPrincipal(
+        token_id="t1", user_id="u1", permission="read",
+        token_type="service", daemon_id="d1",
+    )
+    user = User(id="u1", username="alice", permission="read", is_active=True)
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+    mock_repo.get_user.return_value = user
+    mock_repo.list_broker_credentials.return_value = []
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        resp = client.get("/v1/user_info", headers=_svc_token())
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert "broker" in body
+    assert "brokers" not in body
+
+
+# ---------------------------------------------------------------------------
+# REQ-049 AC-7, REQ-051 AC-7: 복호화 실패 → broker.credential_payload=null
+# ---------------------------------------------------------------------------
+
+def test_user_info_decryption_failure_returns_null_payload() -> None:
+    """REQ-049 AC-7, REQ-051 AC-7: 복호화 실패 → broker.credential_payload=null, 크래시 없음."""
+    settings = _make_settings()
+    principal = TokenPrincipal(
+        token_id="t1", user_id="u1", permission="read",
+        token_type="service", daemon_id="d1",
+    )
+    user = User(id="u1", username="alice", permission="read", is_active=True)
+    # 잘못된 encrypted_payload (복호화 실패)
+    cred = BrokerCredential(
+        user_id="u1", broker="kis", account_ref="ref",
+        encrypted_payload="INVALID_PAYLOAD",
+        credential_version=1, daemon_id="d1",
+    )
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+    mock_repo.get_user.return_value = user
+    mock_repo.list_broker_credentials.return_value = [cred]
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        resp = client.get("/v1/user_info", headers=_svc_token())
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["broker"] is not None
+    assert body["broker"]["account_ref"] == "ref"
+    assert body["broker"]["credential_payload"] is None  # 복호화 실패
+
+
+# ---------------------------------------------------------------------------
+# REQ-051 AC-1: web 토큰 → 403
+# ---------------------------------------------------------------------------
+
+def test_user_info_web_token_returns_403() -> None:
+    """REQ-051 AC-1: web user 토큰(token_type!=service) → 403."""
+    settings = _make_settings()
+    principal = TokenPrincipal(
+        token_id="t1", user_id="u1", permission="read",
+        token_type="user",  # service 아님
+        daemon_id="d1",
+    )
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        resp = client.get("/v1/user_info", headers=_svc_token())
+
+    assert resp.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# REQ-051 AC-2: 비인증 → 401
+# ---------------------------------------------------------------------------
+
+def test_user_info_unauthenticated_returns_401() -> None:
+    """REQ-051 AC-2: Authorization 헤더 없음 → 401."""
+    settings = _make_settings()
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = None  # 토큰 없음
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        resp = client.get("/v1/user_info")  # 헤더 없음
+
+    assert resp.status_code == 401
+
+
+# ---------------------------------------------------------------------------
+# REQ-051 AC-4: 다른 daemon 인증정보 → null (데이터 격리)
+# ---------------------------------------------------------------------------
+
+def test_user_info_different_daemon_credential_not_returned() -> None:
+    """REQ-051 AC-4: daemon_id="d1" token이 daemon_id="d2" 인증정보를 받지 않음."""
+    settings = _make_settings()
+    principal = TokenPrincipal(
+        token_id="t1", user_id="u1", permission="read",
+        token_type="service", daemon_id="d1",
+    )
+    user = User(id="u1", username="alice", permission="read", is_active=True)
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+    mock_repo.get_user.return_value = user
+    # daemon_id="d1" 필터로 조회하면 빈 결과 (DB에는 d2만 있음)
+    mock_repo.list_broker_credentials.return_value = []
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        resp = client.get("/v1/user_info", headers=_svc_token())
+
+    assert resp.status_code == 200
+    assert resp.json()["broker"] is None
+
+
+# ---------------------------------------------------------------------------
+# REQ-051 AC-5: 타 엔드포인트 응답에 credential_payload 미포함
+# ---------------------------------------------------------------------------
+
+def test_other_endpoints_do_not_expose_credential_payload(monkeypatch) -> None:
+    """REQ-051 AC-5: GET /health, /v1/portfolios 등에 credential 필드 없음."""
+    settings = _make_settings()
+    client = TestClient(create_app(settings_factory=lambda: settings))
+
+    # /health 응답 검사
+    resp = client.get("/health")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert "credential_payload" not in body
+    assert "app_secret" not in body
+    assert "app_key" not in body
+
+
+# ---------------------------------------------------------------------------
+# REQ-051 AC-6: DB 오류 → broker=null graceful degrade (500 없음)
+# ---------------------------------------------------------------------------
+
+def test_user_info_db_error_returns_null_broker() -> None:
+    """REQ-051 AC-6: list_broker_credentials DB 오류 → broker=null, 500 없음."""
+    settings = _make_settings()
+    principal = TokenPrincipal(
+        token_id="t1", user_id="u1", permission="read",
+        token_type="service", daemon_id="d1",
+    )
+    user = User(id="u1", username="alice", permission="read", is_active=True)
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+    mock_repo.get_user.return_value = user
+    mock_repo.list_broker_credentials.side_effect = RuntimeError("DB connection error")
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        resp = client.get("/v1/user_info", headers=_svc_token())
+
+    assert resp.status_code == 200
+    assert resp.json()["broker"] is None
+
+
+# ---------------------------------------------------------------------------
+# REQ-051 AC-8: 복호화 실패 시 로그에 app_secret 미노출
+# ---------------------------------------------------------------------------
+
+def test_user_info_decryption_failure_log_does_not_expose_secret(caplog) -> None:
+    """REQ-051 AC-8: 복호화 실패 로그에 app_key, app_secret 값 미포함."""
+    settings = _make_settings()
+    principal = TokenPrincipal(
+        token_id="t1", user_id="u1", permission="read",
+        token_type="service", daemon_id="d1",
+    )
+    user = User(id="u1", username="alice", permission="read", is_active=True)
+    # 잘못된 encrypted_payload
+    cred = BrokerCredential(
+        user_id="u1", broker="kis", account_ref="ref",
+        encrypted_payload="INVALID_PAYLOAD_SECRET_VALUE",
+        credential_version=1, daemon_id="d1",
+    )
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+    mock_repo.get_user.return_value = user
+    mock_repo.list_broker_credentials.return_value = [cred]
+
+    with caplog.at_level(logging.WARNING, logger="app.api.v1_user_info"), \
+         patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        client.get("/v1/user_info", headers=_svc_token())
+
+    # 로그에 app_secret, app_key 값이 없음 확인
+    log_text = " ".join(caplog.messages)
+    assert "mysecret" not in log_text
+    assert "mykey" not in log_text
+    # 로그에 INVALID_PAYLOAD 원문이 없음 (payload 노출 아님)
+    # 단지 오류 유형(class name)만 기록
+    assert "INVALID_PAYLOAD_SECRET_VALUE" not in log_text
+
+
+# ---------------------------------------------------------------------------
+# REQ-049 AC-9: 기존 필드들 회귀 없음
+# ---------------------------------------------------------------------------
+
+def test_user_info_regression_existing_fields_unchanged() -> None:
+    """REQ-049 AC-9: broker 변경이 runtime_id, daemon_id, trading_enabled 등에 영향 없음."""
+    settings = _make_settings()
+    principal = TokenPrincipal(
+        token_id="t1", user_id="u1", permission="read",
+        token_type="service", runtime_id="rt-xyz", daemon_id="d1",
+    )
+    user = User(id="u1", username="alice", email="alice@example.com", permission="read", is_active=True)
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+    mock_repo.get_user.return_value = user
+    mock_repo.list_broker_credentials.return_value = []
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_client(settings, mock_repo)
+        resp = client.get("/v1/user_info", headers=_svc_token())
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["runtime_id"] == "rt-xyz"
+    assert body["daemon_id"] == "d1"
+    assert body["trading_enabled"] is True
+    assert body["user_id"] == "u1"
+    assert body["username"] == "alice"
+    assert body["permission"] == "read"

tests/test_cli.py

@@ -175,6 +175,7 @@ class FakeRepo:
         *,
         account_ref: str,
         encrypted_payload: str,
+        daemon_id: str | None = None,
     ) -> BrokerCredential:
         self.get_user(user_id)
         key = (user_id, broker)
@@ -186,17 +187,65 @@ class FakeRepo:
             encrypted_payload=encrypted_payload,
             credential_version=(existing.credential_version + 1) if existing is not None else 1,
             last_rotated_at=datetime.now(timezone.utc),
+            daemon_id=daemon_id,
         )
         self._broker_credentials[key] = credential
         return credential
 
-    def list_broker_credentials(self, user_id: str) -> list[BrokerCredential]:
-        return sorted(
-            [credential for (owner_user_id, _), credential in self._broker_credentials.items() if owner_user_id == user_id],
-            key=lambda credential: credential.broker,
+    def list_broker_credentials(
+        self,
+        user_id: str,
+        daemon_id: str | None = None,
+    ) -> list[BrokerCredential]:
+        creds = [
+            credential
+            for (owner_user_id, _), credential in self._broker_credentials.items()
+            if owner_user_id == user_id
+        ]
+        if daemon_id is not None:
+            creds = [c for c in creds if c.daemon_id == daemon_id]
+        return sorted(creds, key=lambda credential: credential.broker)
+
+    def update_broker_credential_daemon_id(
+        self,
+        user_id: str,
+        broker: str,
+        daemon_id: str,
+    ) -> bool:
+        key = (user_id, broker)
+        cred = self._broker_credentials.get(key)
+        if cred is None or cred.daemon_id is not None:
+            return False
+        self._broker_credentials[key] = BrokerCredential(
+            user_id=cred.user_id,
+            broker=cred.broker,
+            account_ref=cred.account_ref,
+            encrypted_payload=cred.encrypted_payload,
+            credential_version=cred.credential_version,
+            last_rotated_at=cred.last_rotated_at,
+            daemon_id=daemon_id,
+        )
+        return True
+
+    def get_daemon_by_slug(self, owner_user_id: str, daemon_slug: str):
+        return next(
+            (d for d in self._daemons.values() if d.owner_user_id == owner_user_id and d.daemon_slug == daemon_slug),
+            None,
         )
 
-    def delete_broker_credential(self, user_id: str, broker: str) -> bool:
+    def delete_broker_credential(
+        self,
+        user_id: str,
+        broker: str,
+        daemon_id: str | None = None,
+    ) -> bool:
+        if daemon_id is not None:
+            # daemon_id 기반 삭제
+            for key, cred in list(self._broker_credentials.items()):
+                if cred.daemon_id == daemon_id and cred.broker == broker:
+                    del self._broker_credentials[key]
+                    return True
+            return False
         key = (user_id, broker)
         if key not in self._broker_credentials:
             return False
@@ -986,8 +1035,10 @@ def test_cli_daemon_create_and_list_render_entries() -> None:
 
 
 def test_cli_credential_set_list_and_remove_flow() -> None:
+    """REQ-050: --daemon-id 포함 credential set/list/remove 플로우 (BAT-020 갱신)."""
     repo = FakeRepo(_portfolio())
     user = repo.create_user("alice", permission="read")
+    daemon = repo.create_daemon(user.id, daemon_slug="orb-bot")
     runner = build_runner(repo)
 
     set_result = runner.invoke(
@@ -999,6 +1050,8 @@ def test_cli_credential_set_list_and_remove_flow() -> None:
             "set",
             "--user",
             "alice",
+            "--daemon-id",
+            "orb-bot",
             "--broker",
             "kis",
             "--account-ref",
@@ -1023,18 +1076,16 @@ def test_cli_credential_set_list_and_remove_flow() -> None:
     set_payload = json.loads(set_result.stdout)
     assert set_payload["user_id"] == user.id
     assert set_payload["broker"] == "kis"
+    assert set_payload["daemon_id"] == daemon.id
     assert set_payload["credential_version"] == 1
 
     assert list_result.exit_code == 0
     list_payload = json.loads(list_result.stdout)
-    assert list_payload == [
-        {
-            "broker": "kis",
-            "account_ref": "123-45",
-            "credential_version": 1,
-            "last_rotated_at": list_payload[0]["last_rotated_at"],
-        }
-    ]
+    assert len(list_payload) == 1
+    assert list_payload[0]["broker"] == "kis"
+    assert list_payload[0]["account_ref"] == "123-45"
+    assert list_payload[0]["daemon_id"] == daemon.id
+    assert list_payload[0]["credential_version"] == 1
 
     assert remove_result.exit_code == 0
     remove_payload = json.loads(remove_result.stdout)

tests/test_daemon_assignment.py

@@ -120,7 +120,8 @@ def test_user_info_returns_daemon_id_in_response(monkeypatch) -> None:
     assert resp.json()["daemon_id"] == "d-1"
 
 
-def test_user_info_null_daemon_id_returns_200_with_null(monkeypatch) -> None:
+def test_user_info_null_daemon_id_returns_403(monkeypatch) -> None:
+    """REQ-049 AC-3: daemon_id=None service token → 403 Forbidden (BAT-020 정책 변경)."""
     client = _app_with_db(monkeypatch)
     principal = _principal(daemon_id=None)
     user = User(id="user-1", username="alice", permission="read", is_active=True)
@@ -134,8 +135,8 @@ def test_user_info_null_daemon_id_returns_200_with_null(monkeypatch) -> None:
          patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
         resp = client.get("/v1/user_info", headers=_svc_header())
 
-    assert resp.status_code == 200
-    assert resp.json()["daemon_id"] is None
+    # REQ-049: daemon_id 없는 service token은 user_info 403 (daemon-scoped 인가)
+    assert resp.status_code == 403
 
 
 # ---------------------------------------------------------------------------

tests/test_v1.py

@@ -118,7 +118,7 @@ class _RuntimeIdRoundTripRepo:
             daemon_id=token.daemon_id,
         )
 
-    def list_broker_credentials(self, user_id: str) -> list[BrokerCredential]:
+    def list_broker_credentials(self, user_id: str, daemon_id: str | None = None) -> list[BrokerCredential]:
         return []
 
 
@@ -144,6 +144,9 @@ def test_health_returns_full_response() -> None:
 # ---------------------------------------------------------------------------
 
 def test_auth_issue_service_token_returns_runtime_id(monkeypatch) -> None:
+    # REQ-049: daemon_id 없는 service token → user_info 403
+    # 이 테스트는 service token 발급의 runtime_id 생성을 검증하고,
+    # daemon_id 없는 token으로 user_info 호출 시 403을 확인한다.
     repo = _RuntimeIdRoundTripRepo()
     settings = _make_settings(
         db_enabled=True,
@@ -168,15 +171,13 @@ def test_auth_issue_service_token_returns_runtime_id(monkeypatch) -> None:
         assert issue_body["runtime_id"] is not None
         assert "token" in issue_body
 
+        # REQ-049 AC-3: daemon_id 없는 service token → user_info 403
         user_info_resp = client.get(
             "/v1/user_info",
             headers=_service_token_header(issue_body["token"]),
         )
 
-    assert user_info_resp.status_code == 200
-    user_info_body = user_info_resp.json()
-    assert user_info_body["runtime_id"] == issue_body["runtime_id"]
-    assert user_info_body["username"] == "alice"
+    assert user_info_resp.status_code == 403
 
 
 def test_auth_issue_service_token_generates_distinct_runtime_ids_per_token(monkeypatch) -> None:
@@ -214,11 +215,12 @@ def test_auth_issue_service_token_generates_distinct_runtime_ids_per_token(monke
 # ---------------------------------------------------------------------------
 
 def test_v1_user_info_null_runtime_id_returns_200_with_null(monkeypatch) -> None:
+    # REQ-049: daemon_id 있음 + runtime_id=None → 200 반환
     settings = _make_app_db(monkeypatch)
 
     principal = TokenPrincipal(
         token_id="t1", user_id="user-1", permission="read",
-        token_type="service", runtime_id=None,  # no runtime_id
+        token_type="service", runtime_id=None, daemon_id="daemon-1",
     )
     user = User(id="user-1", username="alice", permission="read", is_active=True)
 
@@ -237,11 +239,12 @@ def test_v1_user_info_null_runtime_id_returns_200_with_null(monkeypatch) -> None
 
 
 def test_v1_user_info_returns_user(monkeypatch) -> None:
+    # REQ-049: daemon_id 있음 + broker=None → 200 반환
     settings = _make_app_db(monkeypatch)
 
     principal = TokenPrincipal(
         token_id="t1", user_id="user-1", permission="read",
-        token_type="service", runtime_id="rt-abc"
+        token_type="service", runtime_id="rt-abc", daemon_id="daemon-1",
     )
     user = User(
         id="user-1", username="alice", email="alice@example.com",
@@ -264,15 +267,39 @@ def test_v1_user_info_returns_user(monkeypatch) -> None:
     assert body["username"] == "alice"
     assert body["trading_enabled"] is True
     assert body["disabled_reason"] is None
-    assert body["brokers"] == []
+    # REQ-049 AC-6: broker 단일 필드 존재, brokers 리스트 미존재
+    assert "broker" in body
+    assert "brokers" not in body
+    assert body["broker"] is None  # 인증정보 없음 → null
+
+
+def test_v1_user_info_no_daemon_id_returns_403(monkeypatch) -> None:
+    """REQ-049 AC-3, REQ-051 AC-3: daemon_id=None 서비스 토큰 → 403."""
+    settings = _make_app_db(monkeypatch)
+
+    principal = TokenPrincipal(
+        token_id="t1", user_id="user-1", permission="read",
+        token_type="service", runtime_id="rt-abc", daemon_id=None,
+    )
+
+    mock_repo = MagicMock()
+    mock_repo.get_token_principal.return_value = principal
+
+    with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
+         patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
+        client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY))
+        resp = client.get("/v1/user_info", headers=_service_token_header())
+
+    assert resp.status_code == 403
 
 
 def test_v1_user_info_inactive_user_has_disabled_reason(monkeypatch) -> None:
+    # daemon_id 있음 + 비활성 user → disabled_reason
     settings = _make_app_db(monkeypatch)
 
     principal = TokenPrincipal(
         token_id="t1", user_id="user-2", permission="read",
-        token_type="service", runtime_id="rt-xyz"
+        token_type="service", runtime_id="rt-xyz", daemon_id="daemon-2",
     )
     user = User(
         id="user-2", username="bob", email=None,
@@ -295,8 +322,8 @@ def test_v1_user_info_inactive_user_has_disabled_reason(monkeypatch) -> None:
     assert body["disabled_reason"] == "account_inactive"
 
 
-def test_v1_user_info_returns_decrypted_broker_credentials(monkeypatch) -> None:
-    """REQ-007 AC5: brokers[]에 복호화된 credential_payload 반환."""
+def test_v1_user_info_returns_decrypted_broker_credential(monkeypatch) -> None:
+    """REQ-049 AC-1: daemon_id 있음 + broker 인증정보 있음 → 단일 broker 반환."""
     from app.core.encryption import encrypt_payload
     import json as _json
 
@@ -307,12 +334,13 @@ def test_v1_user_info_returns_decrypted_broker_credentials(monkeypatch) -> None:
 
     principal = TokenPrincipal(
         token_id="t1", user_id="user-1", permission="read",
-        token_type="service", runtime_id="rt-abc"
+        token_type="service", runtime_id="rt-abc", daemon_id="daemon-1",
     )
     user = User(id="user-1", username="alice", permission="read", is_active=True)
     cred = BrokerCredential(
         user_id="user-1", broker="kis", account_ref="12345678",
         encrypted_payload=encrypted, credential_version=1,
+        daemon_id="daemon-1",
     )
 
     mock_repo = MagicMock()
@@ -331,8 +359,11 @@ def test_v1_user_info_returns_decrypted_broker_credentials(monkeypatch) -> None:
 
     assert resp.status_code == 200
     body = resp.json()
-    assert len(body["brokers"]) == 1
-    broker = body["brokers"][0]
+    # REQ-049 AC-6: broker 단일 필드, brokers 리스트 없음
+    assert "broker" in body
+    assert "brokers" not in body
+    broker = body["broker"]
+    assert broker is not None
     assert broker["broker"] == "kis"
     assert broker["credential_payload"] == {"app_key": "mykey", "app_secret": "mysecret"}
 
@@ -793,12 +824,13 @@ def test_v1_assets_read_token_200(monkeypatch) -> None:
 # GET /v1/user_info
 # ---------------------------------------------------------------------------
 
-def test_v1_user_info_null_runtime_id_returns_200_with_null(monkeypatch) -> None:
+def test_v1_user_info_null_runtime_id_returns_200_with_null_b(monkeypatch) -> None:
+    # REQ-049: daemon_id 있음 + runtime_id=None → 200
     monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true")
 
     principal = TokenPrincipal(
         token_id="t1", user_id="user-1", permission="read",
-        token_type="service", runtime_id=None,  # no runtime_id
+        token_type="service", runtime_id=None, daemon_id="daemon-1",
     )
     user = User(id="user-1", username="alice", permission="read", is_active=True)
 
@@ -816,12 +848,13 @@ def test_v1_user_info_null_runtime_id_returns_200_with_null(monkeypatch) -> None
     assert resp.json()["runtime_id"] is None
 
 
-def test_v1_user_info_returns_user(monkeypatch) -> None:
+def test_v1_user_info_returns_user_b(monkeypatch) -> None:
+    # REQ-049 AC-9 회귀: daemon_id 있음 → 기존 필드들 정상 반환
     monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true")
 
     principal = TokenPrincipal(
         token_id="t1", user_id="user-1", permission="read",
-        token_type="service", runtime_id="rt-abc"
+        token_type="service", runtime_id="rt-abc", daemon_id="daemon-1",
     )
     user = User(
         id="user-1", username="alice", email="alice@example.com",
@@ -831,6 +864,7 @@ def test_v1_user_info_returns_user(monkeypatch) -> None:
     mock_repo = MagicMock()
     mock_repo.get_token_principal.return_value = principal
     mock_repo.get_user.return_value = user
+    mock_repo.list_broker_credentials.return_value = []
 
     with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
          patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):
@@ -842,12 +876,13 @@ def test_v1_user_info_returns_user(monkeypatch) -> None:
     assert body["runtime_id"] == "rt-abc"
     assert body["username"] == "alice"
     assert body["disabled_reason"] is None
+    # REQ-049 AC-6: broker 단일 필드, brokers 키 없음
+    assert "broker" in body
+    assert "brokers" not in body
 
 
 def test_v1_user_info_degrades_when_broker_credentials_lookup_fails(monkeypatch) -> None:
-    """issue #9 regression: broker_credentials lookup failure (missing table / DB error)
-    must NOT crash /v1/user_info with 500. daemon_id/user_id/permission do not depend on
-    broker credentials — degrade gracefully with empty brokers."""
+    """REQ-049 AC-2, REQ-051 AC-6: broker_credentials lookup failure → broker=null graceful degrade."""
     monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true")
 
     principal = TokenPrincipal(
@@ -872,15 +907,18 @@ def test_v1_user_info_degrades_when_broker_credentials_lookup_fails(monkeypatch)
     body = resp.json()
     assert body["daemon_id"] == "daemon-9"
     assert body["user_id"] == "user-1"
-    assert body["brokers"] == []
+    # REQ-049 AC-2: broker=null (graceful degrade, 크래시 없음)
+    assert body["broker"] is None
+    assert "brokers" not in body
 
 
-def test_v1_user_info_inactive_user_has_disabled_reason(monkeypatch) -> None:
+def test_v1_user_info_inactive_user_has_disabled_reason_b(monkeypatch) -> None:
+    # daemon_id 있음 + 비활성 user
     monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true")
 
     principal = TokenPrincipal(
         token_id="t1", user_id="user-2", permission="read",
-        token_type="service", runtime_id="rt-xyz"
+        token_type="service", runtime_id="rt-xyz", daemon_id="daemon-2",
     )
     user = User(
         id="user-2", username="bob", email=None,
@@ -890,6 +928,7 @@ def test_v1_user_info_inactive_user_has_disabled_reason(monkeypatch) -> None:
     mock_repo = MagicMock()
     mock_repo.get_token_principal.return_value = principal
     mock_repo.get_user.return_value = user
+    mock_repo.list_broker_credentials.return_value = []
 
     with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
          patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo):