우리집 비서 만들기 프로젝트 #3-1 - 서버편 첫번째 이야기

fastapi==0.162.0
uvicorn[standard]==0.54.0
pydantic-settings==2.16.2
email-validator==2.3.0
boto3==1.43.123

-r requirements.txt
pytest==9.1.1
httpx==0.28.1

__pycache__/
*.pyc
.pytest_cache/
.venv/
.env
data/

# Cloudflare Tunnel 토큰 (터널 만들 때 받은 eyJ... 값)
TUNNEL_TOKEN=

# Cloudflare Access를 거친 요청만 받기 (cloudflared가 붙이는 JWT 헤더 확인)
SECRETARY_REQUIRE_CF_ACCESS=true

# HMAC 타임스탬프 허용 오차(초)
SECRETARY_CLOCK_SKEW_SECONDS=60

# 승인 대기 유효 시간(초). 지나면 자동 만료
SECRETARY_APPROVAL_TTL_SECONDS=300

# AWS SES (IAM 키는 ses:SendEmail 하나만 허용된 전용 키로)
SECRETARY_SES_REGION=ap-northeast-2
SECRETARY_SES_FROM=Secretary <secretary@example.com>
# 보낼 수 있는 수신자. 쉼표로 구분, "@도메인"은 도메인 전체 허용
# (사용자 등록 때 넣은 본인 이메일로는 이 목록과 상관없이 항상 보낼 수 있다)
SECRETARY_MAIL_ALLOWED_RECIPIENTS=me@example.com,@example.com
AWS_ACCESS_KEY_ID=
AWS_SECRET_ACCESS_KEY=

# 활성 사용자 최대 인원 (한 집 기준)
SECRETARY_MAX_USERS=4

1
from functools import lru_cache
2
from pathlib import Path
3

4
from pydantic_settings import BaseSettings, SettingsConfigDict
5

6

7
class Settings(BaseSettings):
8
    """서버 설정. SECRETARY_ 로 시작하는 환경 변수(또는 .env)에서 값을 읽는다.
9

10
    예: SECRETARY_MAX_USERS=3 → max_users = 3. 환경 변수가 없으면 아래 기본값을 쓴다.
11
    """
12

13
    model_config = SettingsConfigDict(env_prefix="SECRETARY_", env_file=".env", extra="ignore")
14

15
    data_dir: Path = Path("/data")
16

17
    # 한 집에서 쓰는 서비스라 활성 사용자 수에 상한을 둔다
18
    max_users: int = 4
19

20
    # 인증
21
    clock_skew_seconds: int = 60
22
    max_body_bytes: int = 64 * 1024
23
    require_cf_access: bool = True
24
    lockout_threshold: int = 10        # 이 횟수만큼 실패하면
25
    lockout_window_seconds: int = 300  # 이 시간 동안 해당 키를 잠근다
26

27
    # 승인
28
    approval_ttl_seconds: int = 300
29

30
    # 메일 (AWS SES)
31
    ses_region: str = "ap-northeast-2"
32
    ses_from: str = ""
33
    mail_allowed_recipients: str = ""
34

35
    @property
36
    def db_path(self) -> Path:
37
        """SQLite 파일 위치 (Nonce, 인증 실패, 승인 대기열)."""
38
        return self.data_dir / "secretary.db"
39

40
    @property
41
    def registry_path(self) -> Path:
42
        """사용자·기기 저장 파일 위치. 기기 Secret이 들어 있어 권한 600으로 만든다."""
43
        return self.data_dir / "registry.json"
44

45
    @property
46
    def audit_path(self) -> Path:
47
        """감사 로그 파일 위치."""
48
        return self.data_dir / "audit.log"
49

50
    @property
51
    def allowed_recipients(self) -> list[str]:
52
        """쉼표로 이은 수신자 허용 목록을 리스트로 바꾼다.
53

54
        "Dad@Example.com, @family.com" → ["dad@example.com", "@family.com"] (공백 제거, 소문자, 빈 칸 무시)
55
        """
56
        return [r.strip().lower() for r in self.mail_allowed_recipients.split(",") if r.strip()]
57

58

59
@lru_cache
60
def get_settings() -> Settings:
61
    """설정 객체를 돌려준다. @lru_cache 덕분에 처음 한 번만 만들고 그 뒤로는 같은 객체를 재사용한다."""
62
    return Settings()

from functools import lru_cache


@lru_cache
def get_settings():
    print("설정 파일 읽는 중...")
    return {"max_users": 4}


a = get_settings()
b = get_settings()
c = get_settings()
print(a is b is c)

1
"""SQLite 하나에 Nonce 기록, 인증 실패 기록, 승인 대기열을 둔다.
2

3
uvicorn 워커는 1개로 운영한다. 개인 서버라 충분하고, 상태를 한 파일에 모을 수 있다.
4
"""
5
import sqlite3
6
import threading
7
from contextlib import contextmanager
8
from pathlib import Path
9

10
_lock = threading.Lock()
11

12
SCHEMA = """
13
CREATE TABLE IF NOT EXISTS nonces (
14
    access_key TEXT NOT NULL,
15
    nonce      TEXT NOT NULL,
16
    expires_at INTEGER NOT NULL,
17
    PRIMARY KEY (access_key, nonce)
18
);
19
CREATE TABLE IF NOT EXISTS auth_failures (
20
    access_key TEXT NOT NULL,
21
    at         INTEGER NOT NULL
22
);
23
CREATE INDEX IF NOT EXISTS idx_auth_failures ON auth_failures (access_key, at);
24
CREATE TABLE IF NOT EXISTS approvals (
25
    id             TEXT PRIMARY KEY,
26
    action         TEXT NOT NULL,
27
    payload        TEXT NOT NULL,
28
    summary        TEXT NOT NULL,
29
    requested_user TEXT NOT NULL,
30
    requested_by   TEXT NOT NULL,
31
    rule           TEXT NOT NULL,
32
    status         TEXT NOT NULL,
33
    created_at     INTEGER NOT NULL,
34
    expires_at     INTEGER NOT NULL,
35
    decided_by     TEXT,
36
    decided_at     INTEGER,
37
    result         TEXT
38
);
39
CREATE INDEX IF NOT EXISTS idx_approvals_user ON approvals (requested_user, status);
40
"""
41

42

43
class Database:
44
    """SQLite 연결 하나를 감싼 클래스. 모든 읽기/쓰기는 tx()를 거친다."""
45

46
    def __init__(self, path: Path):
47
        """DB 파일을 열고(없으면 만들고) 테이블을 준비한다.
48

49
        check_same_thread=False: 여러 스레드가 이 연결 하나를 같이 쓴다(동시 접근은 _lock으로 막는다).
50
        isolation_level=None: 트랜잭션 시작/끝을 tx()에서 직접 정한다.
51
        WAL: 쓰는 동안에도 읽기가 막히지 않는 저널 방식.
52
        """
53
        path.parent.mkdir(parents=True, exist_ok=True)
54
        self._conn = sqlite3.connect(path, check_same_thread=False, isolation_level=None)
55
        self._conn.row_factory = sqlite3.Row
56
        self._conn.execute("PRAGMA journal_mode=WAL")
57
        self._conn.executescript(SCHEMA)
58

59
    @contextmanager
60
    def tx(self):
61
        """트랜잭션 하나를 연다. with 블록이 무사히 끝나면 COMMIT, 예외가 나면 ROLLBACK.
62

63
            with db.tx() as c:
64
                c.execute("INSERT ...")
65

66
        BEGIN IMMEDIATE는 시작하자마자 쓰기 잠금을 잡아서, "읽고 → 확인하고 → 쓰기" 사이에
67
        다른 요청이 끼어들지 못하게 한다(Nonce 중복 확인, 승인 한 번만 처리에 필요).
68
        """
69
        with _lock:
70
            self._conn.execute("BEGIN IMMEDIATE")
71
            try:
72
                yield self._conn
73
                self._conn.execute("COMMIT")
74
            except Exception:
75
                self._conn.execute("ROLLBACK")
76
                raise
77

78
    def close(self) -> None:
79
        """DB 연결을 닫는다. 서버 종료 때, CLI가 잠깐 열었다 닫을 때 쓴다."""
80
        self._conn.close()

from contextlib import contextmanager


@contextmanager
def tx():
    print("BEGIN")
    try:
        yield "conn"
        print("COMMIT")
    except Exception:
        print("ROLLBACK")
        raise


with tx() as c:
    print("작업:", c)

try:
    with tx() as c:
        raise ValueError("중간에 실패")
except ValueError:
    print("에러는 밖으로 전달됨")

1
"""권한 범위(scope), 역할(role), 기기 프리셋.
2

3
실제로 쓸 수 있는 권한 = 기기에 준 scope AND 사용자 역할이 허용하는 scope.
4
그래서 아이 계정의 기기에 실수로 넓은 프리셋을 줘도 역할 상한을 넘지 못한다.
5
"""
6

7
SYSTEM_READ = "system:read"            # 내 정보, 상태 조회
8
MAIL_REQUEST = "mail:request"          # 메일 발송 요청 (승인 대기열에 올라감)
9
APPROVALS_DECIDE = "approvals:decide"  # 내가 결정할 수 있는 요청 조회, 승인, 거절
10

11
ALL_SCOPES = {SYSTEM_READ, MAIL_REQUEST, APPROVALS_DECIDE}
12

13
# 역할
14
ADMIN = "admin"    # 집 주인. 모든 권한, 아이 요청 승인, 키 관리는 NAS CLI로
15
MEMBER = "member"  # 어른 가족. 자기 요청은 자기 기기로 승인
16
CHILD = "child"    # 아이. 위험한 요청은 관리자가 승인해야 실행
17

18
ROLES = (ADMIN, MEMBER, CHILD)
19

20
ROLE_SCOPES: dict[str, set[str]] = {
21
    ADMIN: set(ALL_SCOPES),
22
    MEMBER: {SYSTEM_READ, MAIL_REQUEST, APPROVALS_DECIDE},
23
    CHILD: {SYSTEM_READ, MAIL_REQUEST},
24
}
25

26
# 기기 종류별 추천 기본값 (CLI에서 --preset으로 사용)
27
PRESETS: dict[str, set[str]] = {
28
    "phone": {SYSTEM_READ, MAIL_REQUEST, APPROVALS_DECIDE},
29
    "watch": {SYSTEM_READ, APPROVALS_DECIDE},
30
    "glasses": {SYSTEM_READ, MAIL_REQUEST},
31
    "mcp": {SYSTEM_READ},
32
    "test": {SYSTEM_READ},
33
}
34

35

36
def effective_scopes(device_scopes, role: str) -> set[str]:
37
    """이 기기로 실제로 쓸 수 있는 권한 = 기기에 준 scope AND 역할이 허용하는 scope.
38

39
    &는 교집합(AND, 양쪽에 모두 있는 것만). 예) 아이(child) 기기에 approvals:decide를 줘도
40
    child 역할 목록에 없으므로 빠진다. 모르는 역할이면 빈 집합(아무 권한 없음).
41
    """
42
    return set(device_scopes) & ROLE_SCOPES.get(role, set())

1
"""사용자와 기기 저장소 (data/registry.json).
2

3
사람(user)이 있고, 기기(device)는 사람에게 속한다. 한 집에서 쓰는 서비스라
4
활성 사용자 수에 상한(기본 4명)을 둔다.
5

6
HMAC은 서버도 같은 비밀값으로 서명을 다시 계산해야 하므로 기기 Secret을 원문으로
7
보관한다. 그래서 파일을 처음부터 권한 600으로 만들고, 이 파일은 데이터 볼륨 밖으로
8
내보내지 않는다.
9
"""
10
import json
11
import os
12
import re
13
import secrets
14
import threading
15
import time
16
from dataclasses import asdict, dataclass
17
from pathlib import Path
18

19
from .scopes import ADMIN, ALL_SCOPES, ROLES
20

21
# 사용자 ID와 기기 이름 규칙. fullmatch()로 쓰기 때문에 문자열 "전체"가 이 모양이어야 한다.
22
# (Access Key 앞부분이 되고 HTTP 헤더로 오가므로 영문 소문자, 숫자, 하이픈만 허용)
23
#   [a-z0-9]         첫 글자: 영문 소문자 또는 숫자 (하이픈으로 시작할 수 없다)
24
#   [a-z0-9-]{1,23}  나머지 1~23자: 영문 소문자, 숫자, 하이픈
25
#   → 전체 2~24자
26
#   통과: dad, kid1, phone-flutter
27
#   거부: -dad (하이픈 시작), Dad (대문자), 내폰 (한글), a (1자), 25자 이상
28
_ID_RE = re.compile(r"[a-z0-9][a-z0-9-]{1,23}")
29

30
# 이메일 형식 확인. 실제로 존재하는 주소인지가 아니라 "이메일처럼 생겼는지"만 가볍게 본다.
31
#   [^@\s]+  '@'도 공백도 아닌 글자 1개 이상 → 아이디 부분 (dad)
32
#   @        '@' 기호 딱 하나
33
#   [^@\s]+  '@'도 공백도 아닌 글자 1개 이상 → 도메인 이름 (example)
34
#   \.       점(.) 하나. 정규식에서 '.'은 "아무 글자"라는 뜻이라 \ 를 붙여 진짜 점으로 쓴다
35
#   [^@\s]+  점 뒤의 나머지 → (com, co.kr)
36
#   ([^...]는 "괄호 안의 글자가 아닌 것", \s는 공백 문자, +는 "1개 이상")
37
#   통과: dad@example.com, mom@mail.co.kr
38
#   거부: dad@example (점 없음), dad@@example.com (@ 두 개), "a b@c.d" (공백)
39
_EMAIL_RE = re.compile(r"[^@\s]+@[^@\s]+\.[^@\s]+")
40

41

42
class RegistryError(ValueError):
43
    """사용자·기기 등록 규칙을 어겼을 때 내는 오류. 메시지는 그대로 CLI 화면에 보여 준다."""
44

45

46
@dataclass
47
class User:
48
    """가족 한 사람. role은 admin/member/child 중 하나, email은 메일 Reply-To에 쓴다."""
49

50
    id: str
51
    name: str
52
    role: str
53
    created_at: int
54
    email: str | None = None
55
    disabled: bool = False
56

57
    def public(self) -> dict:
58
        """API 응답에 내보낼 정보만 고른다 (이메일은 빼고 ID, 이름, 역할만)."""
59
        return {"id": self.id, "name": self.name, "role": self.role}
60

61

62
@dataclass
63
class Device:
64
    """사람에게 속한 기기 하나 (폰, 워치, 안경...). access_key가 기기 ID이고 secret이 서명용 비밀값.
65

66
    폐기해도 지우지 않고 revoked=True로 남긴다. 감사 로그가 가리키는 키를 나중에도 알아볼 수 있게.
67
    """
68

69
    access_key: str
70
    user: str
71
    name: str
72
    secret: str
73
    scopes: list[str]
74
    created_at: int
75
    revoked: bool = False
76
    revoked_at: int | None = None
77

78

79
class Registry:
80
    """사용자와 기기를 registry.json 파일 하나에 저장하고 꺼내 쓰는 저장소."""
81

82
    def __init__(self, path: Path, max_users: int = 4):
83
        """파일 경로와 최대 사용자 수를 받고, 파일이 있으면 바로 읽어 둔다."""
84
        self.path = path
85
        self.max_users = max_users
86
        self._lock = threading.Lock()
87
        self._users: dict[str, User] = {}
88
        self._devices: dict[str, Device] = {}
89
        self._mtime: float | None = None
90
        self._load()
91

92
    def _load(self) -> None:
93
        """registry.json을 읽어 메모리에 올린다. 파일 수정 시각이 바뀌었을 때만 다시 읽는다.
94

95
        CLI(docker exec)는 서버와 다른 프로세스라서, CLI로 기기를 추가하면 파일만 바뀐다.
96
        그래서 매번 수정 시각을 보고, 바뀌었으면 서버를 재시작하지 않아도 새 내용을 읽는다.
97
        """
98
        if not self.path.exists():
99
            self._users, self._devices, self._mtime = {}, {}, None
100
            return
101
        mtime = self.path.stat().st_mtime
102
        if mtime == self._mtime:
103
            return
104
        raw = json.loads(self.path.read_text(encoding="utf-8"))
105
        self._users = {k: User(**v) for k, v in raw.get("users", {}).items()}
106
        self._devices = {k: Device(**v) for k, v in raw.get("devices", {}).items()}
107
        self._mtime = mtime
108

109
    def _save(self) -> None:
110
        """메모리 내용을 registry.json에 쓴다.
111

112
        1. 임시 파일(.tmp)을 처음부터 권한 600(나만 읽고 쓰기)으로 만들어 쓰고
113
        2. os.replace로 원래 파일과 한 번에 바꿔치기한다.
114
        쓰는 도중 서버가 죽어도 원래 파일이 반쯤 쓰인 채로 남지 않는다.
115
        """
116
        self.path.parent.mkdir(parents=True, exist_ok=True)
117
        tmp = self.path.with_suffix(".tmp")
118
        payload = json.dumps(
119
            {
120
                "users": {k: asdict(v) for k, v in self._users.items()},
121
                "devices": {k: asdict(v) for k, v in self._devices.items()},
122
            },
123
            ensure_ascii=False,
124
            indent=2,
125
        )
126
        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
127
        with os.fdopen(fd, "w", encoding="utf-8") as f:
128
            f.write(payload)
129
        os.replace(tmp, self.path)
130
        self._mtime = self.path.stat().st_mtime
131

132
    # ── 사용자 ─────────────────────────────────────────────
133

134
    def user(self, user_id: str) -> User | None:
135
        """ID로 사용자 한 명을 찾는다. 없으면 None."""
136
        with self._lock:
137
            self._load()
138
            return self._users.get(user_id)
139

140
    def users(self) -> list[User]:
141
        """모든 사용자 목록 (비활성 사용자 포함)."""
142
        with self._lock:
143
            self._load()
144
            return list(self._users.values())
145

146
    def add_user(self, user_id: str, name: str, role: str, email: str | None = None) -> User:
147
        """사용자를 추가한다. 규칙에 어긋나면 RegistryError.
148

149
        - ID는 영문 소문자·숫자·하이픈 2~24자, 역할은 admin/member/child 중 하나
150
        - 활성 사용자는 max_users명까지
151
        - 첫 사용자(활성 관리자가 아무도 없을 때)는 admin이어야 한다
152
        """
153
        if not _ID_RE.fullmatch(user_id):
154
            raise RegistryError("사용자 ID는 영문 소문자, 숫자, 하이픈으로 2~24자여야 합니다. 예: dad, mom, kid1")
155
        if role not in ROLES:
156
            raise RegistryError(f"역할은 {', '.join(ROLES)} 중 하나여야 합니다.")
157
        if email and not _EMAIL_RE.fullmatch(email):
158
            raise RegistryError("이메일 형식이 올바르지 않습니다.")
159
        with self._lock:
160
            self._load()
161
            if user_id in self._users:
162
                raise RegistryError("이미 있는 사용자 ID입니다.")
163
            active = [u for u in self._users.values() if not u.disabled]
164
            if len(active) >= self.max_users:
165
                raise RegistryError(f"사용자는 최대 {self.max_users}명까지 등록할 수 있습니다.")
166
            if not any(u.role == ADMIN for u in active) and role != ADMIN:
167
                raise RegistryError("첫 사용자는 관리자(admin)여야 합니다.")
168
            user = User(id=user_id, name=name, role=role, email=email, created_at=int(time.time()))
169
            self._users[user_id] = user
170
            self._save()
171
            return user
172

173
    def disable_user(self, user_id: str) -> int:
174
        """사용자를 비활성화하고 그 사람의 기기 키를 전부 폐기한다. 폐기한 키 개수를 돌려준다."""
175
        with self._lock:
176
            self._load()
177
            user = self._users.get(user_id)
178
            if not user or user.disabled:
179
                raise RegistryError("해당 사용자가 없거나 이미 비활성화됐습니다.")
180
            others = [u for u in self._users.values() if u.id != user_id and not u.disabled and u.role == ADMIN]
181
            if user.role == ADMIN and not others:
182
                raise RegistryError("마지막 관리자는 비활성화할 수 없습니다.")
183
            user.disabled = True
184
            count = self._revoke_where(lambda d: d.user == user_id)
185
            self._save()
186
            return count
187

188
    # ── 기기 ───────────────────────────────────────────────
189

190
    def device(self, access_key: str) -> Device | None:
191
        """Access Key로 기기 하나를 찾는다. 없으면 None. (폐기된 기기도 돌려주니 revoked를 따로 확인한다)"""
192
        with self._lock:
193
            self._load()
194
            return self._devices.get(access_key)
195

196
    def devices(self) -> list[Device]:
197
        """모든 기기 목록 (폐기된 기기 포함)."""
198
        with self._lock:
199
            self._load()
200
            return list(self._devices.values())
201

202
    def add_device(self, user_id: str, name: str, scopes: set[str]) -> Device:
203
        """사용자에게 기기 키를 새로 발급한다.
204

205
        Access Key = "사용자ID-기기이름-무작위8자" (예: dad-phone-1a2b3c4d)
206
        Secret     = 무작위 32바이트 (secrets.token_urlsafe: 추측할 수 없는 난수)
207
        scopes는 저장할 때 그대로 두고, 역할 상한은 요청할 때 effective_scopes()로 깎는다.
208
        """
209
        # 이름이 Access Key 앞부분이 되고 HTTP 헤더로 오가므로 영문 소문자, 숫자, 하이픈만 쓴다.
210
        if not _ID_RE.fullmatch(name):
211
            raise RegistryError("기기 이름은 영문 소문자, 숫자, 하이픈으로 2~24자여야 합니다. 예: phone, glasses")
212
        unknown = scopes - ALL_SCOPES
213
        if unknown:
214
            raise RegistryError(f"알 수 없는 scope: {', '.join(sorted(unknown))}")
215
        with self._lock:
216
            self._load()
217
            user = self._users.get(user_id)
218
            if not user or user.disabled:
219
                raise RegistryError("해당 사용자가 없거나 비활성화됐습니다.")
220
            access_key = f"{user_id}-{name}-{secrets.token_hex(4)}"
221
            device = Device(
222
                access_key=access_key,
223
                user=user_id,
224
                name=name,
225
                secret=secrets.token_urlsafe(32),
226
                scopes=sorted(scopes),
227
                created_at=int(time.time()),
228
            )
229
            self._devices[access_key] = device
230
            self._save()
231
            return device
232

233
    def _revoke_where(self, predicate) -> int:
234
        """조건(predicate)에 맞는 기기를 모두 폐기 표시하고, 폐기한 개수를 돌려준다.
235

236
        predicate는 기기를 받아 True/False를 돌려주는 함수. 예) lambda d: d.user == "kid1"
237
        저장(_save)은 부르는 쪽에서 한다.
238
        """
239
        now = int(time.time())
240
        count = 0
241
        for device in self._devices.values():
242
            if not device.revoked and predicate(device):
243
                device.revoked = True
244
                device.revoked_at = now
245
                count += 1
246
        return count
247

248
    def revoke(self, access_key: str) -> bool:
249
        """기기 키 하나를 폐기한다. 폐기했으면 True, 없거나 이미 폐기된 키면 False."""
250
        with self._lock:
251
            self._load()
252
            count = self._revoke_where(lambda d: d.access_key == access_key)
253
            if count:
254
                self._save()
255
            return bool(count)
256

257
    def revoke_all(self) -> int:
258
        """비상 정지: 모든 기기 키를 한 번에 무효화한다."""
259
        with self._lock:
260
            self._load()
261
            count = self._revoke_where(lambda d: True)
262
            if count:
263
                self._save()
264
            return count

from dataclasses import asdict, dataclass


@dataclass
class User:
    id: str
    name: str
    role: str = "member"


u = User("mom", "엄마")
print(u)
print(asdict(u))
print(u == User("mom", "엄마"))

"""HMAC 요청 서명. 서버와 클라이언트 예제가 같은 함수를 쓴다.

서명할 문자열 = METHOD \\n PATH[?QUERY] \\n TIMESTAMP \\n NONCE \\n SHA256(body) 16진수
서명 = base64url(HMAC-SHA256(secret, 서명할 문자열))
"""
import base64
import hashlib
import hmac
import secrets
import time

HEADER_KEY = "X-Access-Key"
HEADER_TIMESTAMP = "X-Timestamp"
HEADER_NONCE = "X-Nonce"
HEADER_SIGNATURE = "X-Signature"


def canonical_string(method: str, path: str, query: str, timestamp: str, nonce: str, body: bytes) -> str:
    """서명할 문자열을 만든다. 서버와 클라이언트가 글자 하나까지 똑같이 만들어야 서명이 맞는다.

    예) GET /v1/me, 본문 없음 →
        GET
        /v1/me
        1762300000
        3f9c...(nonce)
        e3b0c442...(빈 본문의 SHA-256)
    본문은 통째로 넣지 않고 해시만 넣는다. 본문이 한 글자라도 바뀌면 해시가 달라진다.
    """
    target = f"{path}?{query}" if query else path
    body_hash = hashlib.sha256(body).hexdigest()
    return "\n".join([method.upper(), target, timestamp, nonce, body_hash])


def sign(secret: str, canonical: str) -> str:
    """서명할 문자열을 Secret으로 HMAC-SHA256 서명해 base64url 문자열로 돌려준다.

    base64url은 +, / 대신 -, _ 를 써서 HTTP 헤더에 그대로 넣을 수 있고, 끝의 = 채움 글자는 뗀다.
    """
    digest = hmac.new(secret.encode(), canonical.encode(), hashlib.sha256).digest()
    return base64.urlsafe_b64encode(digest).decode().rstrip("=")


def build_headers(access_key: str, secret: str, method: str, path: str, query: str = "", body: bytes = b"") -> dict:
    """요청 하나에 붙일 인증 헤더 4개를 만든다 (테스트와 예제 클라이언트용).

    타임스탬프는 지금 시각(초), Nonce는 매번 새 무작위 값이라 같은 요청도 매번 서명이 달라진다.
    """
    timestamp = str(int(time.time()))
    nonce = secrets.token_hex(16)
    canonical = canonical_string(method, path, query, timestamp, nonce, body)
    return {
        HEADER_KEY: access_key,
        HEADER_TIMESTAMP: timestamp,
        HEADER_NONCE: nonce,
        HEADER_SIGNATURE: sign(secret, canonical),
    }

"""모든 /v1 요청 앞에서 도는 인증 의존성.

확인 순서:
  1. Cloudflare Access를 거쳐 왔는지 (cloudflared가 붙이는 JWT 헤더)
  2. 헤더 형식, 본문 크기
  3. 키 잠금 여부 (실패가 몰린 키는 잠시 차단)
  4. 기기 존재 여부와 폐기 여부, 기기 주인(사용자)이 활성 상태인지
  5. 타임스탬프 허용 범위
  6. 서명 일치 (상수 시간 비교)
  7. Nonce 재사용 여부
  8. scope (기기 scope AND 사용자 역할이 허용하는 scope)

실패 원인은 감사 로그에만 남기고, 응답은 항상 같은 401 한 가지로 돌려준다.
어디서 막혔는지 알려 주면 공격자에게 힌트가 되기 때문이다.
"""
import hmac
import re
import time
from dataclasses import dataclass

from fastapi import HTTPException, Request, status

from .registry import Device, User
from .scopes import effective_scopes
from .signing import (
    HEADER_KEY,
    HEADER_NONCE,
    HEADER_SIGNATURE,
    HEADER_TIMESTAMP,
    canonical_string,
    sign,
)

_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{3,64}$")
_NONCE_RE = re.compile(r"^[A-Za-z0-9_-]{16,64}$")
_TS_RE = re.compile(r"^\d{9,11}$")

UNAUTHORIZED = HTTPException(status.HTTP_401_UNAUTHORIZED, detail="인증에 실패했습니다.")
FORBIDDEN = HTTPException(status.HTTP_403_FORBIDDEN, detail="권한이 없습니다.")


@dataclass
class Caller:
    """인증을 통과한 호출자. 라우터 함수는 이걸 받아 누가(user), 어떤 기기로(device), 무엇을 할 수 있는지(scopes) 안다."""

    user: User
    device: Device
    scopes: set[str]
    ip: str | None


def _client_ip(request: Request) -> str | None:
    """요청한 쪽의 실제 IP를 찾는다. 감사 로그에 남기는 용도."""
    # Cloudflare를 거친 요청은 이 헤더에 실제 클라이언트 IP가 있다.
    return request.headers.get("Cf-Connecting-Ip") or (request.client.host if request.client else None)


async def authenticate(request: Request) -> Caller:
    """요청 헤더의 서명을 검사해 호출자(Caller)를 돌려준다. 하나라도 틀리면 401.

    검사 순서는 이 파일 맨 위 설명 참고. 비용이 싼 검사(헤더 형식)를 먼저,
    비싼 검사(DB 조회, 서명 계산)를 나중에 한다.
    """
    state = request.app.state
    settings = state.settings
    audit = state.audit
    ip = _client_ip(request)

    def fail(reason: str, key: str | None = None) -> HTTPException:
        """인증 실패를 감사 로그와 실패 기록 테이블에 남기고, 돌려줄 401 예외를 만든다.

        부르는 쪽에서 raise fail("이유") 로 쓴다. 이유는 로그에만 남고 응답에는 나가지 않는다.
        키 형식이 정상일 때만 실패 기록을 쌓는다(이상한 값으로 테이블을 채우지 못하게).
        """
        audit.write("auth.fail", device=key, ip=ip, reason=reason, path=request.url.path)
        if key and _KEY_RE.match(key):
            with state.db.tx() as c:
                c.execute("INSERT INTO auth_failures (access_key, at) VALUES (?, ?)", (key, int(time.time())))
        return UNAUTHORIZED

    if settings.require_cf_access and not request.headers.get("Cf-Access-Jwt-Assertion"):
        raise fail("no_cf_access")

    key = request.headers.get(HEADER_KEY, "")
    ts = request.headers.get(HEADER_TIMESTAMP, "")
    nonce = request.headers.get(HEADER_NONCE, "")
    signature = request.headers.get(HEADER_SIGNATURE, "")
    if not (_KEY_RE.match(key) and _TS_RE.match(ts) and _NONCE_RE.match(nonce) and signature):
        raise fail("bad_headers", key or None)

    body = await request.body()
    if len(body) > settings.max_body_bytes:
        raise fail("body_too_large", key)

    now = int(time.time())
    with state.db.tx() as c:
        c.execute("DELETE FROM auth_failures WHERE at < ?", (now - settings.lockout_window_seconds,))
        failures = c.execute(
            "SELECT COUNT(*) FROM auth_failures WHERE access_key = ? AND at > ?",
            (key, now - settings.lockout_window_seconds),
        ).fetchone()[0]
    if failures >= settings.lockout_threshold:
        audit.write("auth.locked", device=key, ip=ip, path=request.url.path)
        raise UNAUTHORIZED

    device = state.registry.device(key)
    if device is None:
        raise fail("unknown_key", key)
    if device.revoked:
        raise fail("revoked_key", key)
    user = state.registry.user(device.user)
    if user is None or user.disabled:
        raise fail("user_disabled", key)

    if abs(now - int(ts)) > settings.clock_skew_seconds:
        raise fail("timestamp_skew", key)

    canonical = canonical_string(request.method, request.url.path, request.url.query, ts, nonce, body)
    expected = sign(device.secret, canonical)
    if not hmac.compare_digest(expected, signature):
        raise fail("bad_signature", key)

    # Nonce는 타임스탬프 허용 범위의 두 배 동안 기억한다. 그 뒤에는 타임스탬프 검사에서 걸린다.
    with state.db.tx() as c:
        c.execute("DELETE FROM nonces WHERE expires_at < ?", (now,))
        exists = c.execute(
            "SELECT 1 FROM nonces WHERE access_key = ? AND nonce = ?", (key, nonce)
        ).fetchone()
        if not exists:
            c.execute(
                "INSERT INTO nonces (access_key, nonce, expires_at) VALUES (?, ?, ?)",
                (key, nonce, now + settings.clock_skew_seconds * 2),
            )
    if exists:
        raise fail("nonce_replay", key)

    return Caller(user=user, device=device, scopes=effective_scopes(device.scopes, user.role), ip=ip)


def require(scope: str):
    """scope 하나를 받아 "이 권한이 있어야 통과"하는 의존성 함수를 만들어 돌려준다.

        @router.get("/me")
        async def me(caller: Caller = Depends(require(SYSTEM_READ))):

    인증(401)을 먼저 하고, scope가 없으면 403. scope마다 검사 함수를 찍어 내는 함수다.
    """
    async def dependency(request: Request) -> Caller:
        """FastAPI가 요청마다 실제로 부르는 함수. 인증 후 scope를 확인한다."""
        caller = await authenticate(request)
        if scope not in caller.scopes:
            request.app.state.audit.write(
                "auth.forbidden", user=caller.user.id, device=caller.device.access_key, ip=caller.ip,
                scope=scope, path=request.url.path,
            )
            raise FORBIDDEN
        return caller

    return dependency

1
import asyncio
2
import time
3

4

5
async def work(name: str, seconds: float) -> str:
6
    await asyncio.sleep(seconds)  # 기다리는 동안 다른 일을 하러 간다
7
    return f"{name} 완료"
8

9

10
async def main():
11
    start = time.perf_counter()
12
    results = await asyncio.gather(work("메일", 1), work("노트", 1), work("전등", 1))
13
    print(results, f"{time.perf_counter() - start:.1f}초")
14

15

16
asyncio.run(main())

1
from fastapi import APIRouter, Depends
2

3
from ..auth.guard import Caller, require
4
from ..auth.scopes import SYSTEM_READ
5

6
router = APIRouter(prefix="/v1", tags=["system"])
7

8

9
@router.get("/me")
10
async def me(caller: Caller = Depends(require(SYSTEM_READ))):
11
    """GET /v1/me : 지금 이 키가 누구의 어떤 기기이고, 무엇을 할 수 있는지 돌려준다.
12

13
    새 기기를 설정하고 처음 호출해 보는 용도. 200이 오면 서명과 Cloudflare Access 설정이 모두 맞은 것이다.
14
    """
15
    d = caller.device
16
    return {
17
        "user": caller.user.public(),
18
        "device": {"access_key": d.access_key, "name": d.name},
19
        "scopes": sorted(caller.scopes),
20
    }

1
from fastapi import Depends, FastAPI, Header, HTTPException
2
from fastapi.testclient import TestClient
3

4
app = FastAPI()
5

6

7
def who_are_you(x_name: str = Header("")) -> str:
8
    if not x_name:
9
        raise HTTPException(401, "누구세요?")
10
    return x_name
11

12

13
@app.get("/hello")
14
def hello(name: str = Depends(who_are_you)):
15
    return {"hello": name}
16

17

18
client = TestClient(app)
19
print(client.get("/hello").status_code)
20
print(client.get("/hello", headers={"X-Name": "dad"}).json())

1
import logging
2
import uuid
3
from contextlib import asynccontextmanager
4

5
from fastapi import FastAPI, Request
6
from fastapi.exceptions import RequestValidationError
7
from fastapi.responses import JSONResponse
8
from starlette.exceptions import HTTPException
9

10
from .audit import AuditLog
11
from .auth.registry import Registry
12
from .config import Settings, get_settings
13
from .db import Database
14
from .modules.approvals import ApprovalQueue
15
from .modules.mail import MailSender
16
from .routers import approvals, mail, system
17

18
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
19
logger = logging.getLogger("secretary")
20

21

22
def create_app(settings: Settings | None = None, mail_sender=None) -> FastAPI:
23
    """FastAPI 앱을 만든다. 테스트에서는 settings와 가짜 mail_sender를 넣어 따로 만든다.
24

25
    운영 서버는 맨 아래 app = create_app() 을 uvicorn이 불러 쓴다.
26
    """
27
    settings = settings or get_settings()
28

29
    @asynccontextmanager
30
    async def lifespan(app: FastAPI):
31
        """서버가 켜질 때(yield 전) 필요한 객체를 만들어 app.state에 두고, 꺼질 때(yield 뒤) 정리한다.
32

33
        라우터는 request.app.state.db 처럼 꺼내 쓴다.
34
        """
35
        app.state.settings = settings
36
        app.state.db = Database(settings.db_path)
37
        app.state.registry = Registry(settings.registry_path, settings.max_users)
38
        app.state.audit = AuditLog(settings.audit_path)
39
        app.state.approvals = ApprovalQueue(app.state.db, settings.approval_ttl_seconds)
40
        app.state.mail = mail_sender or MailSender(settings.ses_region, settings.ses_from)
41
        app.state.audit.write("server.start")
42
        yield
43
        app.state.db.close()
44

45
    # 운영에서는 문서 페이지를 열지 않는다. 스펙이 필요하면 로컬에서 따로 뽑는다.
46
    app = FastAPI(title="Secretary", lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
47

48
    @app.middleware("http")
49
    async def request_id(request: Request, call_next):
50
        """모든 응답에 X-Request-Id를 붙이는 미들웨어.
51

52
        예상 못 한 오류가 나면 자세한 내용은 서버 로그에만 남기고, 클라이언트에는 같은 ID가 붙은 500만 돌려준다.
53
        사용자가 ID를 알려 주면 로그에서 그 요청을 바로 찾을 수 있다.
54
        """
55
        rid = uuid.uuid4().hex[:12]
56
        try:
57
            response = await call_next(request)
58
        except Exception:
59
            logger.exception("unhandled error rid=%s path=%s", rid, request.url.path)
60
            response = JSONResponse({"code": "INTERNAL_ERROR", "message": "일시적인 오류가 발생했습니다."}, 500)
61
        response.headers["X-Request-Id"] = rid
62
        return response
63

64
    @app.exception_handler(HTTPException)
65
    async def http_error(request: Request, exc: HTTPException):
66
        """HTTP 오류(401, 403, 404 ...)를 {code, message} 모양의 JSON으로 통일한다."""
67
        return JSONResponse({"code": f"HTTP_{exc.status_code}", "message": exc.detail}, exc.status_code)
68

69
    @app.exception_handler(RequestValidationError)
70
    async def validation_error(request: Request, exc: RequestValidationError):
71
        """요청 형식 오류(422)를 {code, message, fields} 모양으로 바꾼다."""
72
        # 어떤 필드가 왜 틀렸는지만 알려 주고, 입력값 자체는 돌려주지 않는다.
73
        fields = [".".join(str(p) for p in e["loc"][1:]) for e in exc.errors()]
74
        return JSONResponse({"code": "INVALID_REQUEST", "message": "요청 형식이 올바르지 않습니다.", "fields": fields}, 422)
75

76
    @app.get("/healthz", include_in_schema=False)
77
    async def healthz():
78
        """GET /healthz : 서버가 살아 있는지. 인증 없이 열려 있고, 컨테이너 상태 확인(healthcheck)이 부른다."""
79
        return {"ok": True}
80

81
    app.include_router(system.router)
82
    app.include_router(mail.router)
83
    app.include_router(approvals.router)
84
    return app
85

86

87
app = create_app()

1
from contextlib import asynccontextmanager
2

3
from fastapi import FastAPI
4
from fastapi.testclient import TestClient
5

6

7
@asynccontextmanager
8
async def lifespan(app: FastAPI):
9
    print("서버 시작: DB 연결")
10
    yield
11
    print("서버 종료: DB 닫기")
12

13

14
app = FastAPI(lifespan=lifespan)
15

16

17
@app.get("/")
18
def root():
19
    print("요청 처리")
20
    return {}
21

22

23
with TestClient(app) as client:
24
    client.get("/")

1
"""사용자·기기 관리 CLI. NAS 안에서만 실행한다(외부 API로는 키를 발급할 수 없다).
2

3
  # 사용자 (최대 SECRETARY_MAX_USERS명, 기본 4명. 첫 사용자는 admin)
4
  docker exec -it secretary python -m app.cli user add dad "아빠" --role admin --email dad@example.com
5
  docker exec -it secretary python -m app.cli user add kid1 "첫째" --role child
6
  docker exec -it secretary python -m app.cli user list
7
  docker exec -it secretary python -m app.cli user disable kid1        # 그 사람의 기기 키도 전부 폐기
8

9
  # 기기
10
  docker exec -it secretary python -m app.cli device add dad phone --preset phone
11
  docker exec -it secretary python -m app.cli device list
12
  docker exec -it secretary python -m app.cli device revoke <access_key>
13
  docker exec -it secretary python -m app.cli device revoke-all       # 비상 정지
14
"""
15
import argparse
16
import sys
17
from datetime import datetime
18

19
from .audit import AuditLog
20
from .auth.registry import Registry, RegistryError
21
from .auth.scopes import ALL_SCOPES, PRESETS, ROLES, effective_scopes
22
from .config import get_settings
23

24

25
def _fmt(ts: int | None) -> str:
26
    """유닉스 시각(초)을 '2026-11-04 21:30' 같은 글자로 바꾼다. 값이 없으면 '-'."""
27
    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M") if ts else "-"
28

29

30
def build_parser() -> argparse.ArgumentParser:
31
    """명령어 구조를 정의한다. user/device 아래에 add, list ... 같은 하위 명령이 붙는다.
32

33
    argparse가 잘못된 입력을 걸러 주고, -h 도움말도 자동으로 만든다.
34
    """
35
    parser = argparse.ArgumentParser(prog="secretary")
36
    sub = parser.add_subparsers(dest="cmd", required=True)
37

38
    usr = sub.add_parser("user", help="사용자 관리").add_subparsers(dest="action", required=True)
39
    ua = usr.add_parser("add", help="사용자 추가")
40
    ua.add_argument("user_id", help="영문 소문자 ID. 예: dad, mom, kid1")
41
    ua.add_argument("name", help="표시 이름. 메일 보낸 사람 이름으로도 쓰인다")
42
    ua.add_argument("--role", choices=ROLES, required=True)
43
    ua.add_argument("--email", help="본인 이메일. 메일 Reply-To, 본인에게 보내기 허용에 쓰인다")
44
    usr.add_parser("list", help="사용자 목록")
45
    ud = usr.add_parser("disable", help="사용자 비활성화 (기기 키도 전부 폐기)")
46
    ud.add_argument("user_id")
47

48
    dev = sub.add_parser("device", help="기기 관리").add_subparsers(dest="action", required=True)
49
    da = dev.add_parser("add", help="기기 키 발급")
50
    da.add_argument("user_id")
51
    da.add_argument("name", help="영문 소문자 기기 이름. 예: phone, watch, glasses")
52
    da.add_argument("--preset", choices=sorted(PRESETS))
53
    da.add_argument("--scope", action="append", default=[], help=f"직접 지정: {', '.join(sorted(ALL_SCOPES))}")
54
    dev.add_parser("list", help="기기 목록")
55
    dr = dev.add_parser("revoke", help="기기 키 폐기")
56
    dr.add_argument("access_key")
57
    dev.add_parser("revoke-all", help="모든 기기 키 폐기 (비상 정지)")
58
    return parser
59

60

61
def main(argv: list[str] | None = None) -> int:
62
    """CLI 시작점. 명령을 해석해 _user 또는 _device로 넘긴다.
63

64
    돌려주는 값은 종료 코드: 0 성공, 1 대상 없음, 2 입력 오류(규칙 위반).
65
    """
66
    args = build_parser().parse_args(argv)
67
    settings = get_settings()
68
    reg = Registry(settings.registry_path, settings.max_users)
69
    audit = AuditLog(settings.audit_path)
70

71
    try:
72
        if args.cmd == "user":
73
            return _user(args, reg, audit)
74
        return _device(args, reg, audit)
75
    except RegistryError as e:
76
        print(e, file=sys.stderr)
77
        return 2
78

79

80
def _user(args, reg: Registry, audit: AuditLog) -> int:
81
    """user add / list / disable 을 처리한다. 바꾼 내용은 감사 로그에 via=cli로 남긴다."""
82
    if args.action == "add":
83
        user = reg.add_user(args.user_id, args.name, args.role, args.email)
84
        audit.write("user.add", user=user.id, via="cli", role=user.role)
85
        print(f"사용자를 추가했습니다: {user.id} ({user.name}, {user.role})")
86
        return 0
87
    if args.action == "list":
88
        users = reg.users()
89
        active = sum(1 for u in users if not u.disabled)
90
        print(f"활성 사용자 {active}/{reg.max_users}명")
91
        for u in users:
92
            state = "비활성" if u.disabled else "활성"
93
            print(f"  {u.id:<12} {u.name:<10} {u.role:<7} {state:<4} {u.email or '-'}")
94
        return 0
95
    if args.action == "disable":
96
        count = reg.disable_user(args.user_id)
97
        audit.write("user.disable", user=args.user_id, via="cli", revoked_devices=count)
98
        print(f"비활성화했습니다. 기기 키 {count}개를 폐기했습니다.")
99
        return 0
100
    return 1
101

102

103
def _device(args, reg: Registry, audit: AuditLog) -> int:
104
    """device add / list / revoke / revoke-all 을 처리한다.
105

106
    add는 Secret을 이때 한 번만 화면에 보여 준다. 역할 상한 때문에 빠지는 scope가 있으면 같이 알려 준다.
107
    """
108
    if args.action == "add":
109
        scopes = set(PRESETS.get(args.preset, set())) | set(args.scope)
110
        if not scopes:
111
            print("--preset 또는 --scope 중 하나는 지정해야 합니다.", file=sys.stderr)
112
            return 2
113
        device = reg.add_device(args.user_id, args.name, scopes)
114
        user = reg.user(args.user_id)
115
        effective = effective_scopes(device.scopes, user.role)
116
        audit.write("device.add", user=user.id, device=device.access_key, via="cli", scopes=device.scopes)
117
        print("기기 키를 발급했습니다. Secret은 지금 한 번만 보여 줍니다.\n")
118
        print(f"  사용자     : {user.id} ({user.name}, {user.role})")
119
        print(f"  Access Key : {device.access_key}")
120
        print(f"  Secret     : {device.secret}")
121
        print(f"  Scopes     : {', '.join(sorted(effective))}")
122
        dropped = set(device.scopes) - effective
123
        if dropped:
124
            print(f"  (역할 {user.role} 상한 때문에 실제로는 쓸 수 없는 scope: {', '.join(sorted(dropped))})")
125
        return 0
126
    if args.action == "list":
127
        for d in reg.devices():
128
            state = "폐기됨" if d.revoked else "사용 중"
129
            print(f"{d.access_key:<32} {d.user:<10} {state:<5} 발급 {_fmt(d.created_at)}  {', '.join(d.scopes)}")
130
        return 0
131
    if args.action == "revoke":
132
        ok = reg.revoke(args.access_key)
133
        if ok:
134
            audit.write("device.revoke", device=args.access_key, via="cli")
135
        print("폐기했습니다." if ok else "해당 키가 없거나 이미 폐기됐습니다.")
136
        return 0 if ok else 1
137
    if args.action == "revoke-all":
138
        count = reg.revoke_all()
139
        audit.write("device.revoke_all", via="cli", count=count)
140
        print(f"{count}개 키를 폐기했습니다.")
141
        return 0
142
    return 1
143

144

145
if __name__ == "__main__":
146
    sys.exit(main())