우리집 비서 만들기 프로젝트 #6 - 서버편: 가족 등록과 테스트

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
  docker exec -it secretary python -m app.cli user delete typo-id      # 기기·승인 기록이 없는 사용자만 완전 삭제
9

10
  # 기기
11
  docker exec -it secretary python -m app.cli device add dad phone --preset phone
12
  docker exec -it secretary python -m app.cli device list
13
  docker exec -it secretary python -m app.cli device revoke <access_key>
14
  docker exec -it secretary python -m app.cli device revoke-all       # 비상 정지
15

16
  # 모듈
17
  docker exec -it secretary python -m app.cli module list             # 연결 상태와 액션(= scope) 목록
18
"""
19
import argparse
20
import sys
21
from datetime import datetime
22

23
import httpx
24

25
from .audit import AuditLog
26
from .auth.registry import Registry, RegistryError
27
from .catalog import Manifest
28
from .auth.scopes import HUB_SCOPES, PRESETS, ROLES, effective_scopes
29
from .config import get_settings
30
from .db import Database
31
from .approvals import ApprovalQueue
32

33

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

38

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

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

47
    usr = sub.add_parser("user", help="사용자 관리").add_subparsers(dest="action", required=True)
48
    ua = usr.add_parser("add", help="사용자 추가")
49
    ua.add_argument("user_id", help="영문 소문자 ID. 예: dad, mom, kid1")
50
    ua.add_argument("name", help="표시 이름. 메일 보낸 사람 이름으로도 쓰인다")
51
    ua.add_argument("--role", choices=ROLES, required=True)
52
    ua.add_argument("--email", help="본인 이메일. 모듈에 함께 전달된다 (메일 모듈은 Reply-To와 본인에게 보내기 허용에 쓴다)")
53
    usr.add_parser("list", help="사용자 목록")
54
    ud = usr.add_parser("disable", help="사용자 비활성화 (기기 키도 전부 폐기)")
55
    ud.add_argument("user_id")
56
    udel = usr.add_parser("delete", help="사용자 완전 삭제 (기기·승인 기록이 없을 때만)")
57
    udel.add_argument("user_id")
58

59
    dev = sub.add_parser("device", help="기기 관리").add_subparsers(dest="action", required=True)
60
    da = dev.add_parser("add", help="기기 키 발급")
61
    da.add_argument("user_id")
62
    da.add_argument("name", help="영문 소문자 기기 이름. 예: phone, watch, glasses")
63
    da.add_argument("--preset", choices=sorted(PRESETS))
64
    da.add_argument("--scope", action="append", default=[], help=f"직접 지정 (여러 번 가능): {', '.join(sorted(HUB_SCOPES))}, <모듈>:<액션>, <모듈>:*, modules:*")
65
    dev.add_parser("list", help="기기 목록")
66
    dr = dev.add_parser("revoke", help="기기 키 폐기")
67
    dr.add_argument("access_key")
68
    dev.add_parser("revoke-all", help="모든 기기 키 폐기 (비상 정지)")
69

70
    mod = sub.add_parser("module", help="모듈 확인").add_subparsers(dest="action", required=True)
71
    mod.add_parser("list", help="등록된 모듈의 연결 상태와 액션 목록")
72
    return parser
73

74

75
def main(argv: list[str] | None = None) -> int:
76
    """CLI 시작점. 명령을 해석해 _user, _device, _module로 넘긴다.
77

78
    돌려주는 값은 종료 코드: 0 성공, 1 대상 없음, 2 입력 오류(규칙 위반).
79
    """
80
    args = build_parser().parse_args(argv)
81
    settings = get_settings()
82
    if args.cmd == "module":
83
        return _module(settings)
84
    reg = Registry(settings.registry_path, settings.max_users)
85
    audit = AuditLog(settings.audit_path)
86

87
    try:
88
        if args.cmd == "user":
89
            return _user(args, reg, audit)
90
        return _device(args, reg, audit)
91
    except RegistryError as e:
92
        print(e, file=sys.stderr)
93
        return 2
94

95

96
def _user(args, reg: Registry, audit: AuditLog) -> int:
97
    """user add / list / disable / delete 를 처리한다. 바꾼 내용은 감사 로그에 via=cli로 남긴다."""
98
    if args.action == "add":
99
        user = reg.add_user(args.user_id, args.name, args.role, args.email)
100
        audit.write("user.add", user=user.id, via="cli", role=user.role)
101
        print(f"사용자를 추가했습니다: {user.id} ({user.name}, {user.role})")
102
        return 0
103
    if args.action == "list":
104
        users = reg.users()
105
        active = sum(1 for u in users if not u.disabled)
106
        print(f"활성 사용자 {active}/{reg.max_users}명")
107
        for u in users:
108
            state = "비활성" if u.disabled else "활성"
109
            print(f"  {u.id:<12} {u.name:<10} {u.role:<7} {state:<4} {u.email or '-'}")
110
        return 0
111
    if args.action == "disable":
112
        count = reg.disable_user(args.user_id)
113
        audit.write("user.disable", user=args.user_id, via="cli", revoked_devices=count)
114
        print(f"비활성화했습니다. 기기 키 {count}개를 폐기했습니다.")
115
        return 0
116
    if args.action == "delete":
117
        settings = get_settings()
118
        db = Database(settings.db_path)
119
        try:
120
            history = ApprovalQueue(db, settings.approval_ttl_seconds).has_history(args.user_id)
121
        finally:
122
            db.close()
123
        user = reg.delete_user(args.user_id, has_history=history)
124
        audit.write("user.delete", user=user.id, via="cli", role=user.role)
125
        print(f"삭제했습니다: {user.id} ({user.name}, {user.role})")
126
        return 0
127
    return 1
128

129

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

133
    add는 Secret을 이때 한 번만 화면에 보여 준다. 역할 상한 때문에 빠지는 scope가 있으면 같이 알려 준다.
134
    """
135
    if args.action == "add":
136
        scopes = set(PRESETS.get(args.preset, set())) | set(args.scope)
137
        if not scopes:
138
            print("--preset 또는 --scope 중 하나는 지정해야 합니다.", file=sys.stderr)
139
            return 2
140
        device = reg.add_device(args.user_id, args.name, scopes)
141
        user = reg.user(args.user_id)
142
        effective = effective_scopes(device.scopes, user.role)
143
        audit.write("device.add", user=user.id, device=device.access_key, via="cli", scopes=device.scopes)
144
        print("기기 키를 발급했습니다. Secret은 지금 한 번만 보여 줍니다.\n")
145
        print(f"  사용자     : {user.id} ({user.name}, {user.role})")
146
        print(f"  Access Key : {device.access_key}")
147
        print(f"  Secret     : {device.secret}")
148
        print(f"  Scopes     : {', '.join(sorted(effective))}")
149
        dropped = set(device.scopes) - effective
150
        if dropped:
151
            print(f"  (역할 {user.role} 상한 때문에 실제로는 쓸 수 없는 scope: {', '.join(sorted(dropped))})")
152
        return 0
153
    if args.action == "list":
154
        for d in reg.devices():
155
            state = "폐기됨" if d.revoked else "사용 중"
156
            print(f"{d.access_key:<32} {d.user:<10} {state:<5} 발급 {_fmt(d.created_at)}  {', '.join(d.scopes)}")
157
        return 0
158
    if args.action == "revoke":
159
        ok = reg.revoke(args.access_key)
160
        if ok:
161
            audit.write("device.revoke", device=args.access_key, via="cli")
162
        print("폐기했습니다." if ok else "해당 키가 없거나 이미 폐기됐습니다.")
163
        return 0 if ok else 1
164
    if args.action == "revoke-all":
165
        count = reg.revoke_all()
166
        audit.write("device.revoke_all", via="cli", count=count)
167
        print(f"{count}개 키를 폐기했습니다.")
168
        return 0
169
    return 1
170

171

172
def _module(settings) -> int:
173
    """module list 를 처리한다. 모듈마다 manifest를 받아 와서 액션과 필요한 scope를 보여 준다.
174

175
    기기에 어떤 --scope 를 줘야 할지 모를 때 이 목록을 보면 된다.
176
    """
177
    endpoints = settings.module_endpoints
178
    if not endpoints:
179
        print("등록된 모듈이 없습니다. SECRETARY_MODULES 를 확인하세요.")
180
        return 0
181
    for name, (url, token) in endpoints.items():
182
        try:
183
            res = httpx.get(f"{url}/manifest", headers={"Authorization": f"Bearer {token}"}, timeout=5)
184
            res.raise_for_status()
185
            manifest = Manifest.model_validate(res.json())
186
        except Exception as e:  # 어떤 이유든 "연결 안 됨"으로 보여 주고 다음 모듈로
187
            print(f"{name:<12} 연결 안 됨  {url}  ({type(e).__name__})")
188
            continue
189
        print(f"{name:<12} 정상        {url}  ({manifest.title})")
190
        for action, spec in manifest.actions.items():
191
            notes = (["승인 필요"] if spec.approval else []) + ([f"역할: {', '.join(spec.roles)}"] if spec.roles else [])
192
            print(f"    {name}:{action:<20} {spec.title}{'  (' + ', '.join(notes) + ')' if notes else ''}")
193
    return 0
194

195

196
if __name__ == "__main__":
197
    sys.exit(main())

import json
import time

import httpx
import pytest
from fastapi.testclient import TestClient

from app.auth.registry import Registry
from app.auth.scopes import ADMIN, CHILD, MEMBER, PRESETS
from app.auth.signing import build_headers
from app.config import Settings
from app.main import create_app
from tests import fake_module

CF = {"Cf-Access-Jwt-Assertion": "test-jwt"}
DOWN_TOKEN = "x" * 40


@pytest.fixture
def env(tmp_path):
    settings = Settings(
        data_dir=tmp_path, require_cf_access=True, lockout_threshold=5,
        modules="demo=http://fake/demo,down=http://fake/down",
        module_tokens=f"demo={fake_module.TOKEN},down={DOWN_TOKEN}",
    )
    reg = Registry(settings.registry_path, settings.max_users)
    reg.add_user("dad", "아빠", ADMIN, "dad@example.com")
    reg.add_user("mom", "엄마", MEMBER, "mom@example.com")
    reg.add_user("kid", "첫째", CHILD)
    d = {
        "dad_phone": reg.add_device("dad", "phone", PRESETS["phone"]),
        "dad_watch": reg.add_device("dad", "watch", PRESETS["watch"]),
        "dad_glasses": reg.add_device("dad", "glasses", PRESETS["glasses"]),
        "mom_phone": reg.add_device("mom", "phone", PRESETS["phone"]),
        "mom_watch": reg.add_device("mom", "watch", PRESETS["watch"]),
        "kid_phone": reg.add_device("kid", "phone", PRESETS["phone"]),  # 역할 상한으로 승인 권한은 잘린다
        "dad_ping": reg.add_device("dad", "ping-only", {"demo:ping"}),
    }
    module_app, executed = fake_module.build()
    app = create_app(settings, module_transport=httpx.ASGITransport(app=module_app))
    with TestClient(app) as client:
        yield client, settings, reg, d, executed


def call(client, device, method, path, body=None, query="", headers=None):
    raw = json.dumps(body).encode() if body is not None else b""
    h = {**CF, **build_headers(device.access_key, device.secret, method, path, query, raw)}
    if body is not None:
        h["Content-Type"] = "application/json"
    h.update(headers or {})
    url = f"{path}?{query}" if query else path
    return client.request(method, url, content=raw, headers=h)


def command(client, device, action="send", module="demo", **params):
    return call(client, device, "POST", "/v1/commands", {"module": module, "action": action, "params": params})


def request_send(client, device, to="grandma@family.example"):
    return command(client, device, "send", to=to)


def pending_ids(client, device):
    r = call(client, device, "GET", "/v1/approvals", query="status=pending")
    assert r.status_code == 200
    return [a["id"] for a in r.json()["items"]]


# ── 인증 ─────────────────────────────────────────────────────

1
#!/usr/bin/env python3
2
"""Secretary 호출 예제 클라이언트. 표준 라이브러리만 쓴다.
3

4
다른 클라이언트(Flutter, G2, Zepp)도 이 파일의 sign_headers()와 같은 방식으로 서명하면 된다.
5

6
환경 변수:
7
  SECRETARY_URL          https://api.example.house
8
  CF_CLIENT_ID           Cloudflare Access 서비스 토큰 ID
9
  CF_CLIENT_SECRET       Cloudflare Access 서비스 토큰 Secret
10
  SECRETARY_ACCESS_KEY   Secretary 기기 키
11
  SECRETARY_SECRET       Secretary 기기 Secret
12

13
사용 예:
14
  python3 secretary_client.py GET /v1/me
15
  python3 secretary_client.py GET /v1/modules
16
  python3 secretary_client.py POST /v1/commands '{"module":"mail","action":"send","params":{"to":["me@example.com"],"subject":"테스트","body":"안녕"}}'
17
  python3 secretary_client.py GET "/v1/approvals?status=pending"
18
  python3 secretary_client.py POST /v1/approvals/<id>/approve
19
"""
20
import base64
21
import hashlib
22
import hmac
23
import json
24
import os
25
import secrets
26
import sys
27
import time
28
import urllib.error
29
import urllib.parse
30
import urllib.request
31

32
USER_AGENT = "secretary-client/1.0"
33

34

35
def sign_headers(access_key: str, secret: str, method: str, path: str, query: str, body: bytes) -> dict:
36
    """요청 하나에 붙일 서명 헤더 4개를 만든다. 서버의 app/auth/signing.py와 똑같은 방식.
37

38
    다른 언어로 클라이언트를 만들 때 이 함수만 옮겨 적으면 된다.
39
    """
40
    timestamp = str(int(time.time()))
41
    nonce = secrets.token_hex(16)
42
    target = f"{path}?{query}" if query else path
43
    canonical = "\n".join([method.upper(), target, timestamp, nonce, hashlib.sha256(body).hexdigest()])
44
    digest = hmac.new(secret.encode(), canonical.encode(), hashlib.sha256).digest()
45
    return {
46
        "X-Access-Key": access_key,
47
        "X-Timestamp": timestamp,
48
        "X-Nonce": nonce,
49
        "X-Signature": base64.urlsafe_b64encode(digest).decode().rstrip("="),
50
    }
51

52

53
def main() -> int:
54
    """명령줄 인자로 받은 요청을 보내고, 상태 코드와 응답 JSON을 보기 좋게 출력한다.
55

56
    Cloudflare Access 서비스 토큰 헤더 + Secretary 서명 헤더를 둘 다 붙인다. 400 이상이면 종료 코드 1.
57
    """
58
    if len(sys.argv) < 3:
59
        print(__doc__)
60
        return 2
61
    method, target = sys.argv[1].upper(), sys.argv[2]
62
    body = sys.argv[3].encode() if len(sys.argv) > 3 else b""
63
    parsed = urllib.parse.urlsplit(target)
64

65
    env = os.environ
66
    headers = {
67
        # urllib 기본 User-Agent(Python-urllib/3.x)는 Cloudflare Browser Integrity Check에 막힌다 (error code: 1010)
68
        "User-Agent": USER_AGENT,
69
        "CF-Access-Client-Id": env["CF_CLIENT_ID"],
70
        "CF-Access-Client-Secret": env["CF_CLIENT_SECRET"],
71
        **sign_headers(env["SECRETARY_ACCESS_KEY"], env["SECRETARY_SECRET"], method, parsed.path, parsed.query, body),
72
    }
73
    if body:
74
        headers["Content-Type"] = "application/json"
75

76
    req = urllib.request.Request(env["SECRETARY_URL"].rstrip("/") + target, data=body or None, method=method,
77
                                 headers=headers)
78
    try:
79
        with urllib.request.urlopen(req, timeout=10) as res:
80
            status, payload = res.status, res.read()
81
    except urllib.error.HTTPError as e:
82
        status, payload = e.code, e.read()
83
    print(status)
84
    try:
85
        print(json.dumps(json.loads(payload), ensure_ascii=False, indent=2))
86
    except ValueError:
87
        print(payload.decode(errors="replace")[:500])
88
    return 0 if status < 400 else 1
89

90

91
if __name__ == "__main__":
92
    sys.exit(main())