diff --git a/.github/workflows/LINT_CHECK.yml b/.github/workflows/LINT_CHECK.yml
index 6455cef..c20ab92 100644
--- a/.github/workflows/LINT_CHECK.yml
+++ b/.github/workflows/LINT_CHECK.yml
@@ -40,3 +40,7 @@ jobs:
- uses: ./.github/actions/setup
- name: Check types
run: pnpm typecheck
+ - name: Check attendance behavior
+ run: pnpm --filter @koin/attendance test
+ - name: Check attendance POSIX runtime
+ run: pnpm --filter @koin/attendance test:runtime
diff --git a/apps/attendance/README.md b/apps/attendance/README.md
index 8e728c2..0cc3d73 100644
--- a/apps/attendance/README.md
+++ b/apps/attendance/README.md
@@ -1,33 +1,72 @@
# 비공개 출근부 스냅샷·자동화 모니터
-`bcsd-attendance`는 Pi가 게시한 출입 현황의 관측 시점 스냅샷을 읽는 Worker입니다. 출입 원장을 저장하거나 회원·태그를 수정하지 않습니다. 오류 수신 요약과 Rocky 헬스는 별도 Worker service binding으로 조회합니다.
+Pi의 Python 리더·LAN 서버가 학번을 Internal 회원과 대조하고 SQLite에 입퇴실을 저장합니다. Worker는 별도 KV 스냅샷을 제공하는 읽기 전용 원격 화면입니다. Worker VPC binding, Basic 비밀번호, UID 수동 등록은 사용하지 않습니다.
+
+```mermaid
+flowchart LR
+ Card["학생증 HCE"] --> Reader["Pi reader.py"]
+ Reader --> LAN["loopback LAN API · SQLite"]
+ LAN --> Publisher["운영자 PC 또는 전용 게시기"]
+ Publisher --> KV["별도 VIEW_STATE KV"]
+ KV --> Viewer["Access · 조직 Gateway로 보호한 원격 화면"]
+```
-## 인증과 읽기 전용 경계
+## 출입 원장과 회원 식별
+
+- 출입 판정은 Pi의 SQLite에서만 수행합니다. 원격 KV의 지연된 읽기로 입퇴실을 판정하지 않습니다.
+- 태그는 입실·퇴실을 번갈아 기록합니다. 중복·오래된 재전송은 새 출입으로 처리하지 않습니다. 최대 6시간을 넘긴 열린 세션은 다음 태그에서 미퇴실 세션으로 닫으며 유효 체류 시간에 합산하지 않습니다.
+- 날짜·이번 주 경계는 KST입니다. 오늘·이번 주 요약은 양수인 유효 체류 시간이 있는 회원의 중앙값입니다.
+- 학번 조회 404는 LAN의 익명 방문자 정책을 따릅니다. 원장에 학번이나 카드 UID를 저장하거나 UID를 학번으로 추측해 연결하지 않습니다.
+- 현재 입실자는 이름·체류 시간만 4열×2행으로 표시하고 넘치는 인원은 페이지를 넘깁니다. 최근 기록·사이드바·수동 카드 등록 패널은 제공하지 않습니다.
+- 16:9 TV 화면의 보이는 글씨는 24px 이상으로 표시합니다. 보조 설명을 줄이고 체류 표는 이름·오늘·이번 주 3열만 사용합니다. 공간이 부족하다고 글씨나 전체 화면 배율을 축소하지 않습니다.
+- 자동 테마는 병천면의 일출·일몰을 브라우저에서 계산합니다. 외부 위치·폰트 서비스를 호출하지 않습니다.
+- LAN은 저장 이벤트로 즉시 갱신하고 KOIN 모니터와 15초 간격으로 교대합니다. 유효 태그는 출근부로 복귀시키고 입실·퇴실 인사 모달을 표시합니다.
+- 실물 학생증의 비금융 학번 읽기 규격이 없으면 지원으로 간주하지 않습니다. 보호 키·금융 앱을 탐색하거나 UID로 사람을 임의 등록하지 않습니다.
+
+## 원격 조회·접근 정책
+
+- 해당 Worker의 모든 URL을 Cloudflare Access로 보호하고 조직 Gateway를 필수로 요구합니다. 일반 소비자 WARP나 클라이언트가 보낸 인증 헤더는 허용 근거가 아닙니다.
+- Worker는 플랫폼의 `ctx.access.aud`를 실제 `CF_ACCESS_AUD`와 비교합니다. 인증되지 않은 별도 호스트·서비스 호출은 데이터를 읽지 못합니다.
+- 회원 자격은 실제 Internal 학번·비밀번호 로그인과 매 요청의 `/v1/members/attendance-session` 검증으로 확인합니다. 이메일 일치, 회원 조회 성공, 별도 Keycloak 계정만으로 로그인 자격을 추정하지 않습니다.
+- Internal의 15분 access token은 `Secure; HttpOnly; SameSite=Strict`의 `__Host-attendance-session` 쿠키에만 보관하며 JSON·브라우저 저장소에 전달하지 않습니다. 계정 비활성화·탈퇴·비밀번호 변경·만료는 다음 검증에서 차단하고 화면의 비공개 상태를 지웁니다.
+- Worker는 고정된 `https://bcsd-attendance-auth.koreatech.in`의 로그인·세션 두 경로만 호출합니다. 전용 Cloudflare 서비스 토큰과 터널의 Access JWT 검증을 사용하며 일반 Internal API·관리자 경로는 공개하지 않습니다.
+- 원격 화면은 기본 180초 간격 스냅샷이며 KV의 지역별 반영 지연이 추가됩니다. 체류 시간·지표는 실제 관측 시각에 고정하고 스냅샷임을 표시합니다. LAN의 즉시 갱신과 다릅니다.
+- 관측 후 5분 초과, 필수 데이터 누락·손상, 인증·연결 실패 시 과거 수치를 현재처럼 연장하지 않고 사용 불가로 처리합니다.
+- `VIEW_STATE`의 `clubroom:view:v1`만 조회합니다. 기존 `ATTENDANCE` KV 데이터는 보존하며 자동으로 병합·이관·삭제하지 않습니다.
+- 서버 응답·관측 시각, 검증된 Wi-Fi PNG, 설정된 실제 Jam PNG만 게시합니다. 카메라·토큰·원본 태그 큐·원장 파일은 게시하지 않습니다.
+
+| Worker 경로 | 용도 |
+| ------------------------------------------------------------ | ----------------------------------------- |
+| `GET /`, `HEAD /` | 실제 LAN 템플릿 기반 원격 화면 |
+| `POST /api/login`, `GET /api/me`, `POST /api/logout` | Internal 로그인·세션 확인·자기 쿠키 종료 |
+| `GET /api/view-status`, `/health` | 스냅샷 가용성·관측 시각 |
+| `GET /api/state` | 출근부 조회 |
+| `GET /api/monitor`, `/api/koin-health`, `/api/worker-health` | 실제 서버·서비스 관측 응답 |
+| `GET /api/spotify`, `/api/jam` | 관측 당시 재생 정보·실제 Jam 설정 여부 |
+| `GET /api/wifi-qr/2.4`, `/api/wifi-qr/5`, `/api/jam-qr` | 스냅샷에 존재하는 검증된 PNG |
+| `GET /api/automation` | Pi 스냅샷과 독립적인 오류 수신·Rocky 상태 |
+
+Worker의 태그·등록·관리자·카메라 쓰기 경로는 제공하지 않습니다. 로그인·로그아웃은 조회 세션만 변경하며 원격 화면에서 출입 원장을 변경할 수 없습니다.
+
+## Worker 빌드·배포
-- 플랫폼이 검증한 `ctx.access.aud`가 운영 `CF_ACCESS_AUD`와 같아야 화면·데이터에 접근할 수 있습니다. 클라이언트가 보낸 JWT 헤더·쿠키·Basic·기기 Bearer를 인증 근거로 해석하지 않습니다.
-- `CF_ACCESS_AUD` 미설정은 503, 플랫폼 Access 컨텍스트 누락·다른 audience는 403입니다. `keep_vars: true`로 운영에서 관리하는 audience를 보존합니다.
-- 허용 목록의 GET과 `/`의 HEAD만 받습니다. 쿼리·해시·관리자·카메라·로그인·쓰기 경로를 중계하지 않습니다. `/api/taps`와 `/api/members`는 제공하지 않습니다.
-- 기존 `DEVICE_TOKEN`·`DASHBOARD_PASSWORD`를 자동 회전·삭제하지 않지만 원격 뷰어의 인증 우회 경로로 사용하지 않습니다. 신규 공개 경로나 Access 권한을 만들지 않습니다.
+```sh
+pnpm --filter @koin/attendance build
+pnpm --filter @koin/attendance typecheck
+pnpm --filter @koin/attendance test
+pnpm --filter @koin/attendance test:runtime
+pnpm deploy:attendance
+```
-## Pi 관측 스냅샷
+Worker 검사는 Node.js 24에서 실행합니다. `test:runtime`은 POSIX 파일 권한을 사용하는 Pi·LAN 코드 때문에 Linux와 Python 3.11 이상이 필요합니다. Windows에서는 Worker 검사와 게시기 검사를 실행하고 리더·LAN 검사는 격리된 Pi/Linux에서 실행합니다. 런타임 검사는 실제 GPIO·회원 API·운영 원장을 사용하지 않습니다.
-`VIEW_STATE`는 기존 조회 전용 KV `b519a98db46d4b088baee97a256bf0f0`의 `clubroom:view:v1`을 읽습니다. 원본 출입 원장 KV를 읽거나 쓰거나 치환하지 않습니다. 게시 작업은 별도 Pi 수집·게시 프로세스의 책임입니다.
+빌드는 설치된 `tsx`로 LAN 렌더러를 실행해 `.wrangler/viewer/dashboard.html`을 생성합니다. Worker는 이 파일을 Text 모듈로 가져오며 별도 Static Assets 인증 우회 경로를 만들지 않습니다.
-스냅샷은 `{version:1, observedAt, publishedAt, payloads, assets}`이며 최대 2 MiB입니다. 관측 시각보다 게시 시각이 빠르거나 미래인 자료, 5분 초과 자료, 누락·잘못된 자료는 503입니다. 실제 LAN origin으로 우회하거나 빈 값으로 바꾸지 않습니다.
+`wrangler.jsonc`에는 실제 별도 KV ID와 service binding을 설정합니다. 운영의 `CF_ACCESS_AUD`는 별도 환경 변수로 관리하고 `keep_vars: true`로 보존합니다. 조직 Gateway·회원 자격 정책이 적용되기 전에는 `workers_dev`를 활성화하지 않습니다. 미리보기 URL은 비활성 상태를 유지하며 원장 KV를 삭제하거나 덮어쓰지 않습니다.
-| 경로 | 응답 |
-| -------------------------------------------------------- | -------------------------------------------- |
-| `/` | 단일 읽기 전용 화면, HEAD 지원 |
-| `/api/state` | 관측 시점의 출입 현황 |
-| `/api/monitor`, `/api/koin-health`, `/api/worker-health` | Pi가 게시한 관측 시점 상태 |
-| `/api/spotify`, `/api/jam` | 관측 시점 재생·참여 정보 |
-| `/api/wifi-qr/2.4`, `/api/wifi-qr/5`, `/api/jam-qr` | 허용된 PNG, 개별 최대 256 KiB |
-| `/api/view-status` | 스냅샷 가용성·관측/게시 시각·나이 |
-| `/health` | 스냅샷 모드·관측/게시 시각·나이 |
-| `/api/events` | 최대 60초 SSE, 10초마다 스냅샷 상태 확인 |
-| `/api/automation` | 독립적인 오류 수신 요약·Rocky 읽기 전용 헬스 |
+`INTERNAL_AUTH_CLIENT_ID`, `INTERNAL_AUTH_CLIENT_SECRET`은 인증 중계 앱 한 곳만 허용하는 전용 서비스 토큰으로 설정합니다. 기존 기기·lookup·monitor·관리자 토큰이나 Internal의 JWT 서명 키를 재사용하지 않습니다. 자격증명·비밀번호·원본 인증 응답을 Git이나 로그에 기록하지 않습니다. 인증 중계가 없거나 검증에 실패하면 데이터를 제공하지 않습니다.
-스냅샷 응답은 게시된 상태 코드를 보존하며 `X-BCSD-Snapshot-*` 헤더로 관측 시각을 구분합니다. 모든 응답은 private/no-store, CSP, nosniff, 프레임 차단을 적용합니다.
+로컬 `wrangler dev`에서는 실제 플랫폼 인증이 없으면 실패합니다. 인증 개발 설정은 로컬 프로세스에만 적용하고 운영에서 클라이언트 헤더로 인증을 흉내 내지 않습니다.
## 오류 수신·Rocky 읽기 전용 모니터
@@ -41,30 +80,29 @@
- 연결 실패·오프라인·숨김·기한 초과는 미확인으로 표시하고 마지막 관측은 과거 기록으로 구분합니다. Pi 스냅샷 표시가 중단돼도 독립 모니터는 표시합니다.
- 오류 전문·프롬프트·소스·로그·원본 provider ID·lease·자격 증명은 반환하지 않습니다. 상세 호스트 설정은 [게이트웨이 운영 계약](../../scripts/autofix/README.md#rocky-읽기-전용-헬스-활성화)을 따릅니다.
-## 화면과 배포
+## 스냅샷 게시기
-Pi 조회 화면·관측 시각·Wi-Fi/Spotify/모니터 화면을 유지하며 외부 UI 라이브러리나 Static Assets를 사용하지 않습니다. `src/dashboard.html`은 Worker Text 모듈입니다. 벽걸이 화면은 패널별 페이지를 사용하고 작은 화면에서는 재배치합니다. 회원 이름 등 데이터는 textContent로만 표시합니다.
+`snapshot/publish.py`는 고정된 Pi5의 허용된 GET 경로만 수집합니다. 기본 실행은 I/O 없는 계획 출력입니다. 실제 account·namespace, 인증 출처, 영구 private checkpoint 디렉터리를 지정하고 `--once` 또는 `--loop`로 게시합니다.
-웹·오류 자동화의 정상 main 승격 후 이 Worker를 별도 배포합니다. `VIEW_STATE` ID와 외부 관리 Access audience, 기존 secret을 유지합니다. secret 설정도 배포 작업입니다. 운영 승인 없이 원장 KV·도메인·Access 정책을 바꾸거나 별도 공개 테스트 Worker를 만들지 않습니다.
+- 게시 간격은 최소 180초, 일일 시도 예산은 600회입니다. 실패한 쓰기도 예산에 포함하며 공유 계정 한도 때문에 무과금을 보장하지 않습니다.
+- 전용 기기 토큰의 KV Storage Write는 계정 범위입니다. 다른 KV에 대한 쓰기 위험을 검토한 뒤 사용합니다.
+- 운영자 PC의 기존 Wrangler 세션은 명시적 `--operator-wrangler-auth-file`로만 사용합니다. 광범위한 OAuth 세션을 Pi로 복사하지 않습니다. 만료 시 인증된 `wrangler kv namespace list`로 정상 갱신하며 `whoami`만으로 갱신됐다고 간주하지 않습니다.
+- 관리자 WARP 경로가 SSH만 허용하면 먼저 고정 호스트 키를 검증하는 `ssh -N -L 127.0.0.1:8080:127.0.0.1:8080 bcsd-pi`를 연결하고 게시기에 `--ssh-forward`를 지정합니다. 연결 대상만 localhost로 바꾸며 원래 Host와 허용 경로를 유지합니다.
+- 게시기·인증·사설 조회 경로가 중단되면 원격 화면도 신선도 제한 후 사용할 수 없습니다. cloudflared 설치 자체는 게시기나 Internal 송신 인증을 대신하지 않습니다.
-```sh
-pnpm --dir apps/attendance exec wrangler types --include-runtime false
-pnpm --dir apps/attendance typecheck
-pnpm --dir apps/attendance test
-pnpm --dir apps/attendance exec wrangler deploy --dry-run
-```
+## LAN 서버
+
+`lan/server.py`는 IPv4 `0.0.0.0:8080`에 바인딩하지만 읽기 출발지는 loopback·`192.168.1.0/24`로 제한합니다. 이 LAN 읽기에는 로그인이 없고 실제 이름·Wi-Fi 비밀번호를 볼 수 있습니다. 인터넷에 직접 게시하지 않습니다.
-로컬에서도 플랫폼 Access 컨텍스트가 없으면 403입니다. 클라이언트 헤더를 인증 우회로 추가하지 않습니다. API 회귀는 격리된 플랫폼 컨텍스트·KV fixture로 검증합니다.
+태그 POST는 loopback 출발지, 올바른 Host·Origin, 별도 `DEVICE_TOKEN`을 모두 요구합니다. 서비스는 root 소유 `0600` 파일의 토큰을 systemd credential로 전달받습니다. `/etc/attendance-lan/config`의 `INTERNAL_LOOKUP_URL`은 고정된 HTTPS `api.internal.bcsdlab.com/v1/service/attendance/member-lookup`이며, lookup·device·monitor 자격증명은 목적별로 분리합니다. HTTPS 전달은 실제 Internal 호스트의 SNI·인증서를 검증하고 TLS 검증을 끄지 않습니다.
-## LAN 리더 배선·설치
+`lan/attendance-lan.service`와 `pi/attendance-reader.service`를 활성화하면 부팅 시 실행됩니다. 실제 사설 회원 조회 경로·토큰을 준비한 뒤 시작합니다. SQLite와 원본 큐는 private 디렉터리·파일 권한을 유지합니다.
-Raspberry Pi 3B 기준입니다. `READER`로 리더 종류를 고릅니다.
-리더의 `ATTENDANCE_URL`은 승인된 LAN 수집 API 주소입니다. 이 읽기 전용 원격 Worker 주소로 태그를 전송하지 않습니다.
+Wi-Fi QR은 실제 비밀번호로 생성한 작은 PNG를 로컬에 설치하고 `WIFI_QR_24_FILE`, `WIFI_QR_5_FILE`로 지정합니다. 화면을 볼 수 있는 사람은 QR에서 비밀번호를 알 수 있으므로 표시 범위를 제한합니다. Jam은 실제 휴대폰 초대 링크로 만든 root 소유 PNG를 `JAM_QR_FILE`로 지정하며 미설정이면 숨깁니다. 곡·기기 URL로 Jam 초대를 만들어 대체하지 않습니다.
-- `rc522`: GPIO(SPI)에 직접 연결한 RC522 모듈. 카드 UID를 16진수 대문자(예: `04A1B2C3`)로 보내고, 결과를 LED 두 개로 알립니다.
-- `hid`(기본값): 카드 UID를 키 입력 + Enter로 보내는 USB HID 리더. 장치를 독점해 UID가 콘솔에 입력되지 않으며, 장치가 여러 개면 `READER_DEVICE`로 경로를 지정합니다.
+## 라즈베리파이 리더
-같은 카드라도 리더 종류에 따라 UID 문자열이 다를 수 있으므로, 리더를 바꾸면 회원을 다시 등록합니다.
+리더는 Android 학생증 HCE에서 읽은 학번·태그 시각을 loopback LAN 서버에만 전송합니다. 서버가 조회한 회원 정보로 입퇴실을 처리하며, 클라우드 UID API·HID 수동 등록은 사용하지 않습니다. 칩 초기화·RF 복구 후에는 카드가 제거된 간격을 확인한 뒤 새 태그를 받으므로 계속 올려 둔 카드를 재시작만으로 새 출입으로 처리하지 않습니다.
### RC522 배선
@@ -84,42 +122,10 @@ Raspberry Pi 3B 기준입니다. `READER`로 리더 종류를 고릅니다.
| 빨강 LED | GPIO → 330 Ω → 애노드, 캐소드 → GND | 18 (GPIO24) |
| LED 공통 GND | GND | 20 |
-| 결과 | LED |
-| --------------------------- | ------------- |
-| 입실 | 초록 1초 |
-| 퇴실 | 초록 2번 깜빡 |
-| 10초 안 중복(무시) | 초록 짧게 1번 |
-| 미등록 카드 | 빨강 1초 |
-| 오프라인·서버 오류(큐 보관) | 빨강 2번 깜빡 |
-| 거절(잘못된 UID 등) | 빨강 3번 깜빡 |
-
-시작할 때 초록·빨강이 차례로 한 번 켜집니다. 카드는 읽은 뒤 HALT시키므로 떼었다 다시 대야 다시 읽힙니다.
-
### 설치
-```sh
-sudo raspi-config nonint do_spi 0 # rc522: 부팅 시 SPI 활성화
-sudo dtparam spi=on # rc522: 재부팅 없이 현재 부팅에도 적용
-sudo apt install python3-spidev python3-rpi-lgpio # Raspberry Pi OS Trixie. hid는 python3-evdev
-sudo install -D -o root -g root -m 755 pi/reader.py /opt/attendance/reader.py
-sudo install -o root -g root -m 644 pi/attendance-reader.service /etc/systemd/system/
-sudo systemctl daemon-reload
-```
-
-RC522는 `/dev/spidev0.0`과 위 배선을 사용합니다. SPI 장치 파일이 있다는 것만으로 모듈 연결이 확인되지는 않습니다. 시작 시 `RC522 응답 없음`이면 3.3V 전원·GND·CE0·SCLK·MOSI·MISO·RST 배선을 확인합니다. USB HID는 `/dev/input/by-id/*-event-kbd`에 실제 리더가 있어야 합니다.
+`reader.py`와 `hce.py`를 함께 `/opt/attendance`에 설치하고 SPI·GPIO 의존성을 준비합니다. APT를 사용할 때는 먼저 해당 Raspberry Pi OS에 맞는 한국 미러를 설정합니다. `/dev/spidev0.0` 존재만으로 카드 통신을 보장하지 않습니다.
-실제 LAN 수집 API 주소와 전용 기기 토큰을 받은 뒤에만 런타임 설정을 작성하고 서비스를 시작합니다. 진단 중이거나 모듈·카드 인식이 확인되지 않았다면 전송 루프를 켜지 않습니다.
-
-```sh
-sudo touch /etc/attendance-reader.env
-sudo chown root:root /etc/attendance-reader.env
-sudo chmod 600 /etc/attendance-reader.env
-sudoedit /etc/attendance-reader.env
-# ATTENDANCE_URL=<실제 LAN 수집 API 주소>
-# DEVICE_TOKEN=
-# READER=rc522 (USB HID는 hid; 필요하면 READER_DEVICE도 지정)
-sudo systemctl enable --now attendance-reader.service
-```
+리더 환경 파일은 root 소유 `0600`으로 관리하고 `ATTENDANCE_URL=http://127.0.0.1:8080`, LAN 서버와 일치하는 `DEVICE_TOKEN`을 지정합니다. 원본 큐 기본 경로는 `/var/lib/attendance/queue.jsonl`이며 `QUEUE_FILE`로 private 경로를 지정할 수 있습니다. 학번·토큰·큐 원문을 Git·로그·공유 문서에 넣지 않습니다.
-- 오프라인이면 태그를 `/var/lib/attendance/queue.jsonl`에 보관하고 30초마다 순서대로 재전송합니다. 태그 시각(`at`)은 파이 시계 기준이므로 NTP 동기화가 필요합니다.
-- 인증 오류(401/403)·408·429·5xx는 큐에 남겨 재시도하고, 그 외 4xx(잘못된 UID 등)는 버립니다.
+서비스는 제한된 private 상태 디렉터리를 사용합니다. 전송 실패는 원래 태그 시각으로 순서대로 재전송하므로 NTP 동기화가 필요합니다. 인증·연결 실패를 성공이나 임의 회원 등록으로 바꾸지 않습니다.
diff --git a/apps/attendance/lan/attendance-lan.service b/apps/attendance/lan/attendance-lan.service
new file mode 100644
index 0000000..719acd3
--- /dev/null
+++ b/apps/attendance/lan/attendance-lan.service
@@ -0,0 +1,30 @@
+[Unit]
+Description=BCSD LAN attendance dashboard and Internal member lookup
+After=network.target
+
+[Service]
+Type=simple
+DynamicUser=yes
+EnvironmentFile=/etc/attendance-lan/config
+LoadCredential=lookup:/etc/attendance-lan/lookup-token
+LoadCredential=device:/etc/attendance-lan/device-token
+Environment=INTERNAL_LOOKUP_TOKEN_FILE=%d/lookup
+Environment=DEVICE_TOKEN_FILE=%d/device
+Environment=ATTENDANCE_HTML=/opt/attendance/lan/dashboard.html
+Environment=ATTENDANCE_DATA_DIR=/var/lib/attendance-lan
+ExecStart=/usr/bin/python3 /opt/attendance/lan/server.py
+StateDirectory=attendance-lan
+StateDirectoryMode=0700
+UMask=0077
+Restart=on-failure
+RestartSec=5
+NoNewPrivileges=yes
+CapabilityBoundingSet=
+ProtectSystem=strict
+ProtectHome=yes
+PrivateTmp=yes
+PrivateDevices=yes
+RestrictAddressFamilies=AF_INET AF_UNIX
+
+[Install]
+WantedBy=multi-user.target
diff --git a/apps/attendance/lan/render-dashboard.mjs b/apps/attendance/lan/render-dashboard.mjs
new file mode 100644
index 0000000..d00ec1e
--- /dev/null
+++ b/apps/attendance/lan/render-dashboard.mjs
@@ -0,0 +1,1105 @@
+#!/usr/bin/env node
+// Render src/dashboard.ts with the tsx loader; only the generated HTML runs on the Pi.
+import { writeFile } from 'node:fs/promises';
+import { resolve } from 'node:path';
+import { pathToFileURL } from 'node:url';
+
+const [modulePath, outputPath] = process.argv.slice(2);
+if (!modulePath || !outputPath)
+ throw new Error('Usage: node --import tsx lan/render-dashboard.mjs ');
+const { DASHBOARD_HTML } = await import(pathToFileURL(resolve(modulePath)).href);
+if (typeof DASHBOARD_HTML !== 'string') throw new Error('Expected the real DASHBOARD_HTML export');
+const startMarker = 'async function checkSession() {';
+const endMarker = "document.getElementById('login-form').addEventListener";
+const start = DASHBOARD_HTML.indexOf(startMarker);
+const end = DASHBOARD_HTML.indexOf(endMarker, start);
+if (start < 0 || end < start || DASHBOARD_HTML.indexOf(startMarker, start + 1) !== -1) {
+ throw new Error('Dashboard authentication structure changed; refusing to generate a partial page');
+}
+const pollMarker = 'setInterval(refresh, POLL_MS);';
+if (
+ DASHBOARD_HTML.indexOf(pollMarker) < 0 ||
+ DASHBOARD_HTML.indexOf(pollMarker) !== DASHBOARD_HTML.lastIndexOf(pollMarker) ||
+ !DASHBOARD_HTML.includes('if (refreshing || !authActive) return;')
+) {
+ throw new Error('Dashboard refresh structure changed; refusing generate partial page');
+}
+const manualRefreshMarker = "document.getElementById('refresh').addEventListener('click', refresh);";
+const headerActionsMarker = '
- 본문으로 건너뛰기
-
-
-
-
- 원격 스냅샷 · 관측 시각 확인 중 · 표시 중단
-
-
-
- 오류 수신 · 24시간
- 확인 중
- Sentry — Cloudflare —
- 수신 확인 —
- 상세
-
-
- Rocky
- 확인 중
- CPU — RAM — 실행기 — 브로커 —
- 호스트 확인 —
- 브로커 HTTP·모델 실행 미확인
-
-
-
- Internal 계정 로그인
- BCSDLab 구성원 계정으로 출근부에 로그인하세요.
-
-
-
-
-
-
-
-
관측 시 동아리방에는
-
등록 회원 확인 중
-
-
— 명
-
-
-
관측 시 입실한 사람
-
동아리방 현황을 불러오고 있습니다.
-
-
-
↗ 첫 입실을 기다리고 있어요
-
모바일 학생증을 리더에 태그해 주세요.
-
-
-
-
-
-
- 회원별 기록을 불러오고 있습니다.
-
-
-
아직 기록이 없어요
-
Internal에 등록된 회원이 모바일 학생증을 태그하면 표시됩니다.
-
-
-
-
Wi-Fi 연결
-
-
-
- QR 코드를 불러오는 중
- BCSD_5G
-
-
-
- QR 코드를 불러오는 중
- BCSD_2.4G
-
-
-
-
-
첫 데이터를 불러오는 중
-
- 아직 동기화되지 않음 · KV 반영 지연 있음 · 관측 시점 기준
-
-
-
-
-
-
KOIN 관측 시점 상태
-
상태 확인 중
-
-
-
- KOIN Workers
- WEB 확인 불가
-
-
- Internal 수집 경로 연결·인증 확인
-
-
-
-
-
-
서버 수집 데이터 대기 · 아직 노드 관측 없음
-
-
-
-
-
-
- 출근부를 보려면 브라우저에서 JavaScript를 활성화해 주세요.
-
-
- BCSD LAB
-
- 5초 후 자동으로 닫힙니다.
- 닫기
-
-
-
- 오류 수신 · Rocky 관측
- 닫기
-
-
- Pi 스냅샷과 독립적인 읽기 전용 관측입니다. CPU는 두 Linux 표본의 차이, RAM은 MemTotal − MemAvailable입니다.
- 서비스·프로세스 상태는 브로커 HTTP 응답·모델 인증·작업 완료의 증거가 아닙니다.
-
- 최근 정상 조회 기록이 없습니다.
-
-
-
-
- `,
+ );
+ replace(
+ '',
+ `
+ .automation-monitor { flex-shrink:0; display:grid; gap:2px; margin:0 0 4px; padding:4px 10px; border:1px solid var(--line); border-radius:16px; background:var(--surface); }
+ .automation-row { display:flex; align-items:center; flex-wrap:wrap; gap:2px 10px; min-width:0; font-size:24px; line-height:1.2; }
+ .automation-row strong { font-size:inherit; }
+ .automation-row small { color:var(--muted); font-size:inherit; }
+ .automation-state { padding:0 6px; border-radius:999px; color:var(--amber); background:var(--amber-container); }
+ .automation-state[data-state="ok"] { color:var(--on-green-container); background:var(--green-container); }
+ .automation-row button { min-height:32px; padding:0 10px; font-size:24px; margin-left:auto; }
+ .automation-dialog { width:min(1000px,calc(100% - 32px)); max-height:85vh; overflow:auto; padding:20px; border:1px solid var(--outline); border-radius:24px; background:var(--surface); color:var(--text); font-size:24px; line-height:1.3; }
+ .automation-dialog::backdrop { background:#10131b88; }
+ .automation-dialog header { display:flex; justify-content:space-between; align-items:center; gap:12px; }
+ .automation-dialog h2 { font-size:28px; margin:0; }
+ .automation-dialog p, .automation-dialog li, .automation-dialog button { font-size:24px; }
+ .automation-dialog ul { padding-left:28px; }
+ @media (min-width:1000px) and (min-aspect-ratio:3/2) {
+ .automation-monitor { padding:4px 12px; gap:3px; }
+ #present .person-name { white-space:normal; overflow:visible; text-overflow:clip; overflow-wrap:anywhere; flex-shrink:0; }
+ #present .elapsed { flex-shrink:0; }
+ }
+ @media (min-width:1000px) and (min-aspect-ratio:3/2) and (max-height:800px) {
+ .app-shell { padding:12px 18px; }
+ .app-header { padding-bottom:6px; margin-bottom:6px; }
+ #cloud-snapshot { margin-bottom:4px; font-size:24px; }
+ .automation-monitor { margin-bottom:6px; }
+ main { grid-template-columns:minmax(0,2fr) minmax(350px,1fr); grid-template-rows:auto minmax(220px,1fr) minmax(180px,1fr) 0; grid-template-areas:"error error" "hero summary" "members wifi" "sync sync"; gap:6px 10px; }
+ .hero { grid-template-columns:140px minmax(0,1fr); }
+ .occupancy, .present-content { padding:8px; }
+ .present-content .section-heading { margin-bottom:4px; }
+ .cards { gap:4px; }
+ .person-card { padding:4px; }
+ .summary-card { padding:4px 8px; }
+ #wifi-section { padding:6px 8px; }
+ .automation-row { gap:2px 8px; }
+ }
+ `,
+ );
+ return html;
+}
diff --git a/apps/attendance/scripts/internal-login.mjs b/apps/attendance/scripts/internal-login.mjs
new file mode 100644
index 0000000..f979e6e
--- /dev/null
+++ b/apps/attendance/scripts/internal-login.mjs
@@ -0,0 +1,445 @@
+// 클라우드 변환에만 삽입한다. LAN 렌더러와 인증 API의 권한 판정은 바꾸지 않는다.
+export const INTERNAL_LOGIN_STYLES = `
+body:not([data-internal-auth="authenticated"]) #main,
+body:not([data-internal-auth="authenticated"]) #monitor-view,
+body:not([data-internal-auth="authenticated"]) #spotify-now,
+body:not([data-internal-auth="authenticated"]) #cloud-snapshot,
+body:not([data-internal-auth="authenticated"]) .automation-monitor,
+body:not([data-internal-auth="authenticated"]) #am-dialog,
+body:not([data-internal-auth="authenticated"]) #greeting-dialog,
+body:not([data-internal-auth="authenticated"]) #viewer-name,
+body:not([data-internal-auth="authenticated"]) #refresh,
+body:not([data-internal-auth="authenticated"]) #logout:not([data-internal-logout-retry]) { display:none !important; }
+body:not([data-internal-auth="authenticated"]) #login-panel { display:block !important; }
+#login-panel { width:min(100%,620px); padding:24px; border-radius:28px; font-size:24px; }
+#login-panel h2 { font-size:clamp(28px,3vw,36px); line-height:1.25; }
+#login-panel p { font-size:24px; line-height:1.3; }
+#login-panel label, #login-panel input, #login-panel button, #viewer-name, #logout { font-size:24px; }
+body[data-internal-auth="authenticated"] #viewer-name { display:inline; }
+#login-panel input, #login-panel button { min-height:52px; }
+#login-panel .internal-session-check { margin-top:12px; }
+#login-panel [aria-busy="true"] { cursor:progress; }
+`;
+
+// 콜백은 동기적으로 실행한다. 중단·메모리 초기화는 기존 Pi/자동화 수명주기가 맡는다.
+// 직렬화 후에도 동작하도록 모듈 바깥의 변수나 import를 참조하지 않는다.
+export function createInternalLogin({ onAuthorized, onUnauthorized, clearPrivateState }) {
+ if (
+ typeof onAuthorized !== 'function' ||
+ typeof onUnauthorized !== 'function' ||
+ typeof clearPrivateState !== 'function'
+ ) {
+ throw new Error('Internal login requires activity and private-state callbacks');
+ }
+ const node = (id) => {
+ const element = document.getElementById(id);
+ if (!element) throw new Error('Internal login markup missing: ' + id);
+ return element;
+ };
+ const panel = node('login-panel');
+ const form = node('login-form');
+ const student = node('login-student');
+ const password = node('login-password');
+ const submit = node('login-submit');
+ const message = node('login-error');
+ const logoutButton = node('logout');
+ const viewerName = node('viewer-name');
+ const main = node('main');
+ const refreshButton = node('refresh');
+ const retry = document.createElement('button');
+ retry.type = 'button';
+ retry.className = 'button-tonal internal-session-check';
+ retry.textContent = '세션 다시 확인';
+ form.appendChild(retry);
+ node('login-heading').textContent = 'Internal 계정 로그인';
+ const description = panel.querySelector('p');
+ if (description && description !== message) description.remove();
+ student.setAttribute('inputmode', 'numeric');
+ student.setAttribute('pattern', '[0-9]{8,10}');
+ student.setAttribute('autocomplete', 'username');
+ password.setAttribute('autocomplete', 'current-password');
+ password.type = 'password';
+ student.required = password.required = true;
+ let running = false;
+ let pageHidden = false;
+ let session = null;
+ let deadline = 0;
+ let revision = 0;
+ let pending = null;
+ let expiryTimer = null;
+ let checkTimer = null;
+ let logoutRequired = false;
+ const visible = () => !pageHidden && !document.hidden && navigator.onLine !== false;
+
+ function paintBusy() {
+ const busy = pending !== null;
+ form.setAttribute('aria-busy', String(busy));
+ submit.disabled = busy || !running || !visible();
+ retry.disabled = busy || !running || !visible() || logoutRequired;
+ student.disabled = password.disabled = busy;
+ logoutButton.disabled = busy || !running || !visible();
+ }
+ function clearPrivateDom() {
+ for (const id of ['present', 'members', 'monitor-cards', 'koin-services', 'am-recent']) {
+ const element = document.getElementById(id);
+ if (element) element.replaceChildren();
+ if (element && (id === 'monitor-cards' || id === 'koin-services')) {
+ const placeholder = document.createElement('p');
+ placeholder.className = 'monitor-waiting';
+ placeholder.textContent = '인증 후 확인';
+ element.appendChild(placeholder);
+ }
+ }
+ for (const id of ['count', 'today-total', 'week-total', 'unrecorded-total', 'members-badge']) {
+ const element = document.getElementById(id);
+ if (element) element.textContent = '—';
+ }
+ for (const id of [
+ 'member-count',
+ 'last-updated',
+ 'spotify-title',
+ 'spotify-artists',
+ 'am-note',
+ 'greeting-message',
+ ]) {
+ const element = document.getElementById(id);
+ if (element) element.textContent = '';
+ }
+ for (const selector of ['#main img', '#monitor-view img', '#spotify-now img', '#greeting-dialog img']) {
+ for (const image of document.querySelectorAll(selector)) {
+ image.removeAttribute('src');
+ image.removeAttribute('srcset');
+ image.hidden = true;
+ }
+ }
+ for (const id of [
+ 'am-ingress-state',
+ 'am-rocky-state',
+ 'am-ingress-time',
+ 'am-rocky-time',
+ 'am-sentry',
+ 'am-cloudflare',
+ 'am-sentry-count',
+ 'am-cloudflare-count',
+ 'am-cpu',
+ 'am-memory',
+ 'am-executor',
+ 'am-broker',
+ 'monitor-timestamp',
+ 'monitor-status',
+ ]) {
+ const element = document.getElementById(id);
+ if (element) {
+ element.textContent = '인증 후 확인';
+ element.dataset.state = 'unknown';
+ }
+ }
+ for (const id of ['am-dialog', 'greeting-dialog']) {
+ const dialog = document.getElementById(id);
+ if (dialog && dialog.open && typeof dialog.close === 'function') dialog.close();
+ }
+ for (const selector of ['#main [data-since]', '#main [data-member-id]']) {
+ for (const element of document.querySelectorAll(selector)) {
+ element.removeAttribute('data-since');
+ element.removeAttribute('data-member-id');
+ }
+ }
+ const monitor = document.getElementById('monitor-view');
+ if (monitor) monitor.hidden = true;
+ }
+ function lockView(text) {
+ document.body.dataset.internalAuth = logoutRequired ? 'logout-required' : 'signed-out';
+ main.hidden = true;
+ panel.hidden = false;
+ viewerName.hidden = refreshButton.hidden = true;
+ viewerName.textContent = '';
+ logoutButton.hidden = !logoutRequired;
+ logoutButton.toggleAttribute('data-internal-logout-retry', logoutRequired);
+ logoutButton.textContent = logoutRequired ? '로그아웃 다시 시도' : '로그아웃';
+ password.value = '';
+ message.textContent = text || '';
+ }
+ function cancelRequest() {
+ const request = pending;
+ pending = null;
+ if (request) {
+ clearTimeout(request.timer);
+ request.controller.abort();
+ }
+ }
+ function deny(text) {
+ revision += 1;
+ session = null;
+ deadline = 0;
+ clearTimeout(expiryTimer);
+ clearTimeout(checkTimer);
+ expiryTimer = checkTimer = null;
+ cancelRequest();
+ lockView(text);
+ try {
+ onUnauthorized();
+ } finally {
+ clearPrivateDom();
+ clearPrivateState();
+ paintBusy();
+ }
+ }
+ function isAuthorized() {
+ if (!session) return false;
+ if (!running || !visible() || performance.now() >= deadline) {
+ deny(visible() ? '세션이 만료됐습니다. 다시 로그인해 주세요.' : '화면으로 돌아오면 계정을 다시 확인합니다.');
+ return false;
+ }
+ return true;
+ }
+ function handleUnauthorized(responseOrStatus) {
+ const status =
+ typeof responseOrStatus === 'number' ? responseOrStatus : responseOrStatus && responseOrStatus.status;
+ if (status !== 401 && status !== 403) return false;
+ deny(status === 403 ? '이 계정의 접근 권한을 확인할 수 없습니다.' : '세션이 만료됐습니다. 다시 로그인해 주세요.');
+ return true;
+ }
+ function scheduleChecks() {
+ clearTimeout(expiryTimer);
+ clearTimeout(checkTimer);
+ expiryTimer = setTimeout(
+ () => {
+ if (session && performance.now() < deadline) scheduleChecks();
+ else deny('세션이 만료됐습니다. 다시 로그인해 주세요.');
+ },
+ Math.max(0, deadline - performance.now()),
+ );
+ checkTimer = setTimeout(checkSession, Math.min(60000, Math.max(0, deadline - performance.now())));
+ }
+ function accept(data, startedAt) {
+ if (
+ !data ||
+ data.authenticated !== true ||
+ typeof data.memberId !== 'string' ||
+ !data.memberId.trim() ||
+ typeof data.name !== 'string' ||
+ !data.name.trim() ||
+ !Number.isSafeInteger(data.servedAt) ||
+ data.servedAt <= 0 ||
+ !Number.isSafeInteger(data.expiresAt) ||
+ data.expiresAt <= data.servedAt ||
+ data.expiresAt - data.servedAt > 900000
+ ) {
+ deny('계정 유효기간을 확인할 수 없습니다. 다시 로그인해 주세요.');
+ return false;
+ }
+ // 응답 본문을 받는 데 걸린 시간도 세션 잔여 시간에서 차감한다.
+ const nextDeadline = startedAt + data.expiresAt - data.servedAt;
+ if (!running || !visible() || performance.now() >= nextDeadline) {
+ deny('세션을 다시 확인해 주세요.');
+ return false;
+ }
+ const resume = session === null;
+ if (session && session.memberId !== data.memberId) {
+ deny('계정이 변경됐습니다. 다시 로그인해 주세요.');
+ return false;
+ }
+ session = {
+ authenticated: true,
+ memberId: data.memberId,
+ name: data.name,
+ expiresAt: data.expiresAt,
+ servedAt: data.servedAt,
+ };
+ deadline = nextDeadline;
+ logoutRequired = false;
+ document.body.dataset.internalAuth = 'authenticated';
+ panel.hidden = true;
+ main.hidden = false;
+ viewerName.textContent = data.name;
+ viewerName.hidden = logoutButton.hidden = refreshButton.hidden = false;
+ logoutButton.removeAttribute('data-internal-logout-retry');
+ logoutButton.textContent = '로그아웃';
+ message.textContent = '';
+ scheduleChecks();
+ if (resume) {
+ try {
+ onAuthorized({ ...session });
+ } catch {
+ deny('화면을 시작할 수 없습니다. 세션을 다시 확인해 주세요.');
+ return false;
+ }
+ }
+ return true;
+ }
+ async function requestAuth(kind, body) {
+ const paths = { session: '/api/me', login: '/api/login', logout: '/api/logout' };
+ const controller = new AbortController();
+ const request = { controller, timer: null };
+ const currentRevision = revision;
+ const startedAt = performance.now();
+ pending = request;
+ request.timer = setTimeout(() => controller.abort(), 8000);
+ paintBusy();
+ try {
+ const response = await fetch(paths[kind], {
+ method: kind === 'session' ? 'GET' : 'POST',
+ credentials: 'same-origin',
+ cache: 'no-store',
+ redirect: 'error',
+ headers:
+ kind === 'session'
+ ? { accept: 'application/json' }
+ : { 'content-type': 'application/json', accept: 'application/json' },
+ body: kind === 'logout' ? '{}' : body,
+ signal: controller.signal,
+ });
+ const text = await response.text();
+ if (pending !== request || revision !== currentRevision || !running) return null;
+ if (controller.signal.aborted) throw new Error('Authentication request timed out');
+ return { status: response.status, data: response.ok && kind !== 'logout' ? JSON.parse(text) : null, startedAt };
+ } catch (error) {
+ if (pending !== request || revision !== currentRevision || !running) return null;
+ throw error;
+ } finally {
+ clearTimeout(request.timer);
+ if (pending === request) pending = null;
+ paintBusy();
+ }
+ }
+ async function checkSession() {
+ if (!running || pending || logoutRequired) return false;
+ if (!visible()) {
+ deny(
+ navigator.onLine === false
+ ? '오프라인입니다. 연결 후 계정을 다시 확인합니다.'
+ : '화면으로 돌아오면 계정을 다시 확인합니다.',
+ );
+ return false;
+ }
+ try {
+ const result = await requestAuth('session');
+ if (!result) return false;
+ if (result.status === 401) {
+ deny('');
+ return false;
+ }
+ if (result.status !== 200) {
+ deny(
+ result.status === 403
+ ? '이 계정의 접근 권한을 확인할 수 없습니다.'
+ : '인증 서비스를 사용할 수 없습니다. 세션을 다시 확인해 주세요.',
+ );
+ return false;
+ }
+ return accept(result.data, result.startedAt);
+ } catch {
+ if (!pending && running && visible()) deny('계정 확인에 실패했습니다. 연결 후 세션을 다시 확인해 주세요.');
+ return false;
+ }
+ }
+ async function login(event) {
+ event.preventDefault();
+ if (!running || pending || session || !visible()) {
+ password.value = '';
+ return;
+ }
+ if (!/^[0-9]{8,10}$/.test(student.value.trim()) || !password.value.trim()) {
+ password.value = '';
+ message.textContent = '8~10자리 학번과 비밀번호를 입력해 주세요.';
+ return;
+ }
+ message.textContent = '로그인 확인 중입니다.';
+ try {
+ // 비밀번호는 이 요청 본문 외에는 보관하지 않으며 자동 재전송하지 않는다.
+ const resultPromise = requestAuth(
+ 'login',
+ JSON.stringify({ studentNumber: student.value.trim(), password: password.value }),
+ );
+ password.value = '';
+ const result = await resultPromise;
+ if (!result) return;
+ if (result.status === 401) {
+ deny('학번 또는 비밀번호를 확인해 주세요.');
+ } else if (result.status === 403) {
+ deny('이 계정의 접근 권한을 확인할 수 없습니다.');
+ } else if (result.status !== 200) {
+ deny('로그인 서비스를 사용할 수 없습니다. 잠시 뒤 직접 다시 시도해 주세요.');
+ } else {
+ accept(result.data, result.startedAt);
+ }
+ } catch {
+ if (!pending && running && visible())
+ deny('로그인 응답을 확인할 수 없습니다. 연결 후 세션을 확인하거나 직접 다시 로그인해 주세요.');
+ } finally {
+ password.value = '';
+ }
+ }
+ async function logout() {
+ if (!running) return false;
+ logoutRequired = true;
+ deny('로그아웃 요청 중입니다.');
+ student.value = '';
+ try {
+ const result = await requestAuth('logout');
+ if (!result) return false;
+ if (result.status < 200 || result.status >= 300) throw new Error('Logout unavailable');
+ logoutRequired = false;
+ deny('로그아웃했습니다. 출퇴근 기록은 변경되지 않습니다.');
+ return true;
+ } catch {
+ if (!pending && running) deny('로그아웃을 완료하지 못했습니다. 연결 후 로그아웃을 다시 시도해 주세요.');
+ return false;
+ }
+ }
+ function suspend() {
+ deny(
+ logoutRequired
+ ? '로그아웃을 완료하려면 연결 후 로그아웃을 다시 시도해 주세요.'
+ : navigator.onLine === false
+ ? '오프라인입니다. 연결 후 계정을 다시 확인합니다.'
+ : '화면으로 돌아오면 계정을 다시 확인합니다.',
+ );
+ }
+ function resume() {
+ if (!running) return;
+ if (!visible()) suspend();
+ else {
+ paintBusy();
+ checkSession();
+ }
+ }
+ function pagehide() {
+ pageHidden = true;
+ suspend();
+ }
+ function pageshow() {
+ pageHidden = false;
+ resume();
+ }
+ function start() {
+ if (running) return;
+ running = true;
+ form.addEventListener('submit', login);
+ retry.addEventListener('click', checkSession);
+ logoutButton.addEventListener('click', logout);
+ document.addEventListener('visibilitychange', resume);
+ window.addEventListener('offline', suspend);
+ window.addEventListener('online', resume);
+ window.addEventListener('pagehide', pagehide);
+ window.addEventListener('pageshow', pageshow);
+ deny('계정 확인 중입니다.');
+ checkSession();
+ }
+ function stop() {
+ running = false;
+ form.removeEventListener('submit', login);
+ retry.removeEventListener('click', checkSession);
+ logoutButton.removeEventListener('click', logout);
+ document.removeEventListener('visibilitychange', resume);
+ window.removeEventListener('offline', suspend);
+ window.removeEventListener('online', resume);
+ window.removeEventListener('pagehide', pagehide);
+ window.removeEventListener('pageshow', pageshow);
+ deny('');
+ }
+ lockView('');
+ clearPrivateDom();
+ paintBusy();
+ return { start, stop, isAuthorized, handleUnauthorized, checkSession, logout };
+}
diff --git a/apps/attendance/snapshot/publish.py b/apps/attendance/snapshot/publish.py
new file mode 100644
index 0000000..18878ef
--- /dev/null
+++ b/apps/attendance/snapshot/publish.py
@@ -0,0 +1,681 @@
+#!/usr/bin/env python3
+"""허용된 LAN 조회 응답을 별도 Cloudflare KV에 게시하는 표준 라이브러리 CLI.
+
+기본 실행은 I/O 없는 계획 출력이며 --once/--loop가 실제 게시를 시작한다.
+--validate는 네트워크·파일 쓰기 없이 명시한 인증과 설정만 검사한다.
+기기 토큰은 목적에 맞게 제한하고, 계정 전체에 적용되는 KV 쓰기 권한의
+위험을 검토한다. 운영자 Wrangler 세션은 명시적인 PC 전용 모드로만 읽으며
+Pi에 복사하지 않는다. 만료 시 인증된 `wrangler kv namespace list`로
+정상 갱신한다. `whoami` 출력만으로 토큰이 갱신됐다고 간주하지 않는다.
+
+checkpoint는 영구 private 디렉터리에 보관한다. 쓰기 전에 시도 예산을
+원자적으로 예약하므로 모호한 실패도 일일 예산에 포함하고 성공 시각은
+진행시키지 않는다. 이 예산은 공유 계정의 무료 한도나 과금을 보장하지 않는다.
+"""
+
+import argparse
+import base64
+from contextlib import contextmanager
+from datetime import datetime, timezone
+import hashlib
+import http.client
+import json
+import math
+import os
+from pathlib import Path
+import re
+import signal
+import ssl
+import stat
+import sys
+import tempfile
+import threading
+import time
+import tomllib
+import zlib
+
+ORIGIN_HOST = "192.168.1.112"
+ORIGIN_PORT = 8080
+JSON_PATHS = ("/api/state", "/api/monitor", "/api/koin-health", "/api/worker-health", "/api/spotify", "/api/jam")
+WIFI_PATHS = ("/api/wifi-qr/2.4", "/api/wifi-qr/5")
+ASSET_PATHS = (*WIFI_PATHS, "/api/jam-qr")
+READ_PATHS = frozenset((*JSON_PATHS, *ASSET_PATHS))
+SNAPSHOT_KEY = "clubroom:view:v1"
+MAX_SNAPSHOT = 2 * 1024 * 1024
+KV_LIMIT = 25 * 1024 * 1024
+MAX_PNG = 65536
+MAX_REPLY = 65536
+MAX_SAFE = 9007199254740991
+DAILY_BUDGET = 600
+DEFAULT_INTERVAL = 180
+ID_PATTERN = re.compile(r"[a-fA-F0-9]{32}\Z")
+TOKEN_PATTERN = re.compile(r"[A-Za-z0-9._~+/=-]{20,8192}\Z")
+
+
+class PublisherError(Exception):
+ """자격증명·원문이 없는 고정 오류 메시지만 CLI 경계를 통과한다."""
+
+
+def now_ms():
+ return time.time_ns() // 1_000_000
+
+
+def pairs_unique(pairs):
+ result = {}
+ for key, value in pairs:
+ if key in result:
+ raise ValueError("duplicate key")
+ result[key] = value
+ return result
+
+
+def parse_json(raw):
+ try:
+ return json.loads(raw.decode("utf-8"), object_pairs_hook=pairs_unique,
+ parse_constant=lambda _: (_ for _ in ()).throw(ValueError("nonfinite")))
+ except (ValueError, UnicodeError, RecursionError):
+ raise PublisherError("invalid JSON response") from None
+
+
+def encode_json(value):
+ try:
+ return json.dumps(value, ensure_ascii=False, allow_nan=False, separators=(",", ":")).encode("utf-8")
+ except (ValueError, UnicodeError, RecursionError):
+ raise PublisherError("invalid snapshot encoding") from None
+
+
+def exact(value, keys):
+ return isinstance(value, dict) and set(value) == set(keys.split())
+
+
+def integer(value, minimum=0):
+ return type(value) is int and minimum <= value <= MAX_SAFE
+
+
+def timestamp(value):
+ return value is None or integer(value)
+
+
+def text(value, limit=256):
+ return (isinstance(value, str) and len(value) <= limit
+ and not any(ord(c) < 32 or 0x7f <= ord(c) < 0xa0 or 0xd800 <= ord(c) <= 0xdfff for c in value))
+
+
+def valid_cover(value):
+ if value is None:
+ return True
+ if not isinstance(value, str) or len(value) > 512:
+ return False
+ return re.fullmatch(r"https://i\.scdn\.co(?::443)?/image/[A-Za-z0-9]{1,128}", value) is not None
+
+
+def valid_state(body):
+ if not exact(body, "now members unknownCards recent") or not integer(body["now"], 1):
+ return False
+ if body["unknownCards"] != [] or not isinstance(body["members"], list) or not isinstance(body["recent"], list):
+ return False
+ for member in body["members"]:
+ if (not exact(member, "name present since todayMs weekMs unrecordedWeek firstInToday lastCheckOut")
+ or not text(member["name"], 100) or not member["name"] or type(member["present"]) is not bool
+ or not all(timestamp(member[key]) for key in ("since", "firstInToday", "lastCheckOut"))
+ or not all(integer(member[key]) for key in ("todayMs", "weekMs", "unrecordedWeek"))):
+ return False
+ for event in body["recent"]:
+ if (not exact(event, "name checkIn checkOut unrecorded") or not text(event["name"], 100)
+ or not integer(event["checkIn"], 1) or not timestamp(event["checkOut"])
+ or type(event["unrecorded"]) is not bool):
+ return False
+ return True
+
+
+def valid_optional(path, body):
+ # 비밀 필드가 섞인 응답은 거부하고 허용된 응답·원본 시각은 그대로 유지한다.
+ if path == "/api/jam":
+ return exact(body, "available") and type(body["available"]) is bool
+ if path == "/api/spotify":
+ return (exact(body, "state title artists updatedAt coverUrl")
+ and body["state"] in ("idle", "loading", "playing", "paused", "error", "unknown")
+ and text(body["title"]) and isinstance(body["artists"], list) and len(body["artists"]) <= 8
+ and all(text(artist, 128) for artist in body["artists"])
+ and timestamp(body["updatedAt"]) and valid_cover(body["coverUrl"]))
+ if path == "/api/monitor":
+ if not exact(body, "observedAt nodes") or not integer(body["observedAt"], 1) or not isinstance(body["nodes"], list):
+ return False
+ seen = set()
+ for node in body["nodes"]:
+ if (not exact(node, "id label collectedAt cpuPercent memoryUsedBytes memoryTotalBytes health")
+ or node["id"] not in ("prod", "stage", "db") or node["id"] in seen
+ or not text(node["label"], 100) or node["health"] not in ("up", "down", "unknown")
+ or not timestamp(node["collectedAt"])
+ or not all(timestamp(node[key]) for key in ("memoryUsedBytes", "memoryTotalBytes"))):
+ return False
+ cpu = node["cpuPercent"]
+ if cpu is not None and (type(cpu) not in (int, float) or not math.isfinite(cpu) or not 0 <= cpu <= 100):
+ return False
+ seen.add(node["id"])
+ return seen == {"prod", "stage", "db"}
+ if path == "/api/koin-health":
+ if not exact(body, "observedAt services") or not integer(body["observedAt"], 1) or not isinstance(body["services"], list):
+ return False
+ rows, fields, identities = body["services"], "id label checkedAt status reason httpStatus", {"prod-api", "stage-api"}
+ elif path == "/api/worker-health":
+ if not exact(body, "checkedAt workers") or not integer(body["checkedAt"], 1) or not isinstance(body["workers"], list):
+ return False
+ rows, fields, identities = body["workers"], "id label status httpStatus reason", {"koin-web"}
+ else:
+ return False
+ seen = set()
+ for row in rows:
+ if (not exact(row, fields) or row["id"] not in identities or row["id"] in seen
+ or not text(row["label"], 100) or row["status"] not in ("up", "down", "unknown")
+ or row["reason"] not in ("unavailable", "http", "invalid-response", "ready", "not-ready", "tls", "timeout", "redirect", "restricted")
+ or (row["httpStatus"] is not None and (type(row["httpStatus"]) is not int or not 100 <= row["httpStatus"] <= 599))
+ or ("checkedAt" in row and not integer(row["checkedAt"], 1))):
+ return False
+ seen.add(row["id"])
+ return seen == identities
+
+
+def valid_png(raw):
+ """제한된 PNG 청크·CRC·비인터레이스 스캔라인을 검사한다."""
+ if not 8 < len(raw) <= MAX_PNG or raw[:8] != b"\x89PNG\r\n\x1a\n":
+ return False
+ offset, header, palette, ended = 8, None, False, False
+ image = bytearray()
+ while offset + 12 <= len(raw):
+ size = int.from_bytes(raw[offset:offset + 4], "big")
+ kind = raw[offset + 4:offset + 8]
+ end = offset + size + 12
+ if (end > len(raw) or not all(65 <= c <= 90 or 97 <= c <= 122 for c in kind)
+ or zlib.crc32(raw[offset + 4:end - 4]) != int.from_bytes(raw[end - 4:end], "big")):
+ return False
+ data = raw[offset + 8:end - 4]
+ if header is None:
+ if kind != b"IHDR" or size != 13:
+ return False
+ width, height = int.from_bytes(data[:4], "big"), int.from_bytes(data[4:8], "big")
+ depth, color, compression, filtering, interlace = data[8:]
+ if (not 1 <= width <= 2048 or not 1 <= height <= 2048
+ or depth not in {0: (1, 2, 4, 8, 16), 2: (8, 16), 3: (1, 2, 4, 8), 4: (8, 16), 6: (8, 16)}.get(color, ())
+ or compression or filtering or interlace):
+ return False
+ row_bytes = (width * {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}[color] * depth + 7) // 8
+ expected = (row_bytes + 1) * height
+ if expected > 16 * 1024 * 1024:
+ return False
+ header = (color, depth, row_bytes, expected)
+ elif kind == b"IHDR":
+ return False
+ elif kind == b"PLTE":
+ if palette or image or color in (0, 4) or not size or size % 3 or size > 768 or (color == 3 and size // 3 > 2 ** depth):
+ return False
+ palette = True
+ elif kind == b"IDAT":
+ if ended or (color == 3 and not palette):
+ return False
+ image.extend(data)
+ elif kind == b"IEND":
+ if size or end != len(raw) or not image:
+ return False
+ try:
+ decoder = zlib.decompressobj()
+ decoded = decoder.decompress(image, expected + 1)
+ return (len(decoded) == expected and decoder.eof and not decoder.unused_data
+ and not decoder.unconsumed_tail
+ and all(decoded[index] <= 4 for index in range(0, expected, row_bytes + 1)))
+ except zlib.error:
+ return False
+ elif not kind[0] & 0x20:
+ return False
+ if image and kind != b"IDAT":
+ ended = True
+ offset = end
+ return False
+
+
+def fetch_lan(path, *, ssh_forward=False):
+ if path not in READ_PATHS:
+ raise PublisherError("LAN target is not allowed")
+ limit = MAX_SNAPSHOT if path == "/api/state" else MAX_PNG if path in ASSET_PATHS else MAX_REPLY
+ # SSH 전용 관리자 경로에서는 명시한 loopback 전달을 사용하되 Host는 유지한다.
+ connection = http.client.HTTPConnection("127.0.0.1" if ssh_forward else ORIGIN_HOST,
+ ORIGIN_PORT, timeout=12)
+ try:
+ connection.request("GET", path, headers={"Host": f"{ORIGIN_HOST}:{ORIGIN_PORT}",
+ "Accept": "image/png" if path in ASSET_PATHS else "application/json", "Cache-Control": "no-store"})
+ response = connection.getresponse()
+ # 리디렉션이나 HTTP_PROXY 환경 변수로 목적지를 바꾸지 않는다.
+ status, mime = response.status, response.getheader("Content-Type", "").split(";", 1)[0].strip().lower()
+ if status != 200:
+ return status, mime, b"" # 서버 오류 원문은 읽거나 전달하지 않는다.
+ raw = response.read(limit + 1)
+ if len(raw) > limit:
+ raise PublisherError("LAN response exceeds size bound")
+ return status, mime, raw
+ except (OSError, http.client.HTTPException):
+ raise PublisherError("LAN request unavailable") from None
+ finally:
+ connection.close()
+
+
+def collect_snapshot(fetch=fetch_lan, clock=now_ms):
+ payloads = {}
+ for path in JSON_PATHS:
+ try:
+ status, mime, raw = fetch(path)
+ if status != 200:
+ if path == "/api/state":
+ raise PublisherError("attendance core unavailable")
+ payloads[path] = {"status": status if type(status) is int and 400 <= status <= 599 else 502,
+ "body": {"error": "source_unavailable"}}
+ continue
+ body = parse_json(raw) if mime == "application/json" else None
+ valid = valid_state(body) if path == "/api/state" else valid_optional(path, body)
+ if not valid:
+ raise PublisherError("invalid source schema")
+ payloads[path] = {"status": 200, "body": body}
+ except (PublisherError, TypeError, KeyError, ValueError, RecursionError):
+ if path == "/api/state":
+ raise PublisherError("attendance core unavailable or invalid") from None
+ payloads[path] = {"status": 502, "body": {"error": "source_unavailable"}}
+ assets = {}
+ jam = payloads["/api/jam"]
+ paths = (*WIFI_PATHS, "/api/jam-qr") if jam["status"] == 200 and jam["body"]["available"] else WIFI_PATHS
+ for path in paths:
+ try:
+ status, mime, raw = fetch(path)
+ if status != 200 or mime != "image/png" or not valid_png(raw):
+ raise PublisherError("required QR image unavailable or invalid")
+ except PublisherError:
+ if path in WIFI_PATHS:
+ raise
+ # 실제 QR이 없으면 Jam을 사용 가능으로 광고하지 않는다.
+ payloads["/api/jam"] = {"status": 502, "body": {"error": "source_unavailable"}}
+ continue
+ assets[path] = base64.b64encode(raw).decode("ascii")
+ observed = payloads["/api/state"]["body"]["now"]
+ published = clock()
+ if observed > published or published - observed > 60_000:
+ raise PublisherError("attendance observation is stale or in the future")
+ snapshot = {"version": 1, "observedAt": observed, "publishedAt": published, "payloads": payloads, "assets": assets}
+ encoded = encode_json(snapshot)
+ if len(encoded) > min(MAX_SNAPSHOT, KV_LIMIT):
+ raise PublisherError("snapshot exceeds 2 MiB bound")
+ return snapshot, encoded
+
+
+def windows_security(path, apply=False):
+ """Windows에서는 chmod 대신 실제 소유자·DACL을 검사한다."""
+ import ctypes
+ from ctypes import wintypes as w
+ adv = ctypes.WinDLL("advapi32", use_last_error=True)
+ kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+ adv.OpenProcessToken.argtypes = [w.HANDLE, w.DWORD, ctypes.POINTER(w.HANDLE)]
+ adv.GetTokenInformation.argtypes = [w.HANDLE, ctypes.c_int, w.LPVOID, w.DWORD, ctypes.POINTER(w.DWORD)]
+ adv.ConvertSidToStringSidW.argtypes = [w.LPVOID, ctypes.POINTER(w.LPWSTR)]
+ adv.ConvertStringSecurityDescriptorToSecurityDescriptorW.argtypes = [w.LPCWSTR, w.DWORD, ctypes.POINTER(w.LPVOID), ctypes.POINTER(w.DWORD)]
+ adv.SetFileSecurityW.argtypes = [w.LPCWSTR, w.DWORD, w.LPVOID]
+ adv.GetNamedSecurityInfoW.argtypes = [w.LPWSTR, ctypes.c_int, w.DWORD, ctypes.POINTER(w.LPVOID), ctypes.POINTER(w.LPVOID), ctypes.POINTER(w.LPVOID), ctypes.POINTER(w.LPVOID), ctypes.POINTER(w.LPVOID)]
+ adv.GetAce.argtypes = [w.LPVOID, w.DWORD, ctypes.POINTER(w.LPVOID)]
+ kernel.GetCurrentProcess.restype = w.HANDLE
+ kernel.CloseHandle.argtypes = [w.HANDLE]
+ kernel.LocalFree.argtypes = [w.LPVOID]
+ def sid_string(sid):
+ result = w.LPWSTR()
+ if not adv.ConvertSidToStringSidW(sid, ctypes.byref(result)):
+ raise PublisherError("cannot inspect Windows private ACL")
+ try:
+ return result.value
+ finally:
+ kernel.LocalFree(ctypes.cast(result, w.LPVOID))
+ token, needed = w.HANDLE(), w.DWORD()
+ if not adv.OpenProcessToken(kernel.GetCurrentProcess(), 8, ctypes.byref(token)):
+ raise PublisherError("cannot inspect Windows identity")
+ try:
+ adv.GetTokenInformation(token, 1, None, 0, ctypes.byref(needed))
+ buffer = ctypes.create_string_buffer(needed.value)
+ if not adv.GetTokenInformation(token, 1, buffer, needed, ctypes.byref(needed)):
+ raise PublisherError("cannot inspect Windows identity")
+ current_sid = sid_string(ctypes.cast(buffer, ctypes.POINTER(w.LPVOID))[0])
+ finally:
+ kernel.CloseHandle(token)
+ descriptor = w.LPVOID()
+ if apply:
+ # 하위 항목에만 상속하며 Users·Everyone 권한은 허용하지 않는다.
+ sddl = f"D:P(A;OICI;FA;;;{current_sid})(A;OICI;FA;;;SY)(A;OICI;FA;;;BA)"
+ if not adv.ConvertStringSecurityDescriptorToSecurityDescriptorW(sddl, 1, ctypes.byref(descriptor), None):
+ raise PublisherError("cannot create Windows private ACL")
+ try:
+ if not adv.SetFileSecurityW(str(path), 0x80000004, descriptor):
+ raise PublisherError("cannot apply Windows private ACL")
+ finally:
+ kernel.LocalFree(descriptor)
+ owner, acl, descriptor = w.LPVOID(), w.LPVOID(), w.LPVOID()
+ if adv.GetNamedSecurityInfoW(str(path), 1, 5, ctypes.byref(owner), None, ctypes.byref(acl), None, ctypes.byref(descriptor)):
+ raise PublisherError("cannot inspect Windows private ACL")
+ try:
+ allowed = {current_sid, "S-1-5-18", "S-1-5-32-544"}
+ if not owner or sid_string(owner) not in allowed or not acl:
+ raise PublisherError("Windows file must have a private owner and DACL")
+ # ACL 헤더의 ACE 개수 필드를 읽는다.
+ count = ctypes.c_ushort.from_address(acl.value + 4).value
+ for index in range(count):
+ ace = w.LPVOID()
+ if not adv.GetAce(acl, index, ctypes.byref(ace)):
+ raise PublisherError("cannot inspect Windows private ACL")
+ kind = ctypes.c_ubyte.from_address(ace.value).value
+ if kind == 1: # A deny ACE cannot grant access.
+ continue
+ if kind != 0 or sid_string(ace.value + 8) not in allowed:
+ raise PublisherError("Windows file grants access outside its private principals")
+ finally:
+ kernel.LocalFree(descriptor)
+
+
+def private_path(path, directory=False):
+ try:
+ info = path.lstat()
+ if (path.is_symlink() or getattr(info, "st_file_attributes", 0) & 0x400
+ or not (stat.S_ISDIR(info.st_mode) if directory else stat.S_ISREG(info.st_mode))
+ or (not directory and info.st_nlink != 1)):
+ raise PublisherError("private path must be a regular non-linked file or directory")
+ if os.name == "nt":
+ windows_security(path)
+ elif info.st_uid not in (0, os.geteuid()) or info.st_mode & 0o077:
+ raise PublisherError("private path must be owner-only (0700 directory / 0600 file)")
+ except OSError:
+ raise PublisherError("private path is inaccessible") from None
+
+
+def private_read(path, limit):
+ private_path(path)
+ try:
+ flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
+ before = path.lstat()
+ with os.fdopen(os.open(path, flags), "rb") as stream:
+ actual = os.fstat(stream.fileno())
+ if (before.st_dev, before.st_ino) != (actual.st_dev, actual.st_ino) or actual.st_size > limit:
+ raise PublisherError("private file changed or exceeds size bound")
+ raw = stream.read(limit + 1)
+ if len(raw) > limit:
+ raise PublisherError("private file exceeds size bound")
+ return raw
+ except OSError:
+ raise PublisherError("private file is inaccessible") from None
+
+
+def read_credential(args, clock=now_ms):
+ try:
+ if args.operator_wrangler_auth_file:
+ data = tomllib.loads(private_read(Path(args.operator_wrangler_auth_file), 65536).decode("utf-8"))
+ value = data.get("oauth_token")
+ expiry = data.get("expiration_time")
+ expiry = datetime.fromisoformat(expiry.replace("Z", "+00:00")) if isinstance(expiry, str) else expiry
+ if not isinstance(expiry, datetime) or expiry.tzinfo is None or expiry.timestamp() * 1000 <= clock() + 60_000:
+ raise PublisherError("operator OAuth expired or missing expiry; run authenticated Wrangler kv namespace list on the authorized PC")
+ elif args.token_file:
+ value = private_read(Path(args.token_file), 8193).decode("ascii").strip()
+ else:
+ value = os.environ.get(args.token_env, "")
+ if not isinstance(value, str) or not TOKEN_PATTERN.fullmatch(value):
+ raise PublisherError("credential is missing or invalid")
+ return value
+ except (OSError, ValueError, UnicodeError, TypeError, OverflowError):
+ raise PublisherError("credential file is missing or invalid") from None
+
+
+def destination(account, namespace):
+ if not isinstance(account, str) or not ID_PATTERN.fullmatch(account) or not isinstance(namespace, str) or not ID_PATTERN.fullmatch(namespace):
+ raise PublisherError("explicit 32-hex account and existing dedicated namespace are required")
+ return f"/client/v4/accounts/{account.lower()}/storage/kv/namespaces/{namespace.lower()}/values/{SNAPSHOT_KEY}"
+
+
+def put_snapshot(account, namespace, credential, body):
+ path = destination(account, namespace)
+ if not isinstance(body, bytes) or not 0 < len(body) <= min(MAX_SNAPSHOT, KV_LIMIT):
+ raise PublisherError("snapshot exceeds write bound")
+ connection = http.client.HTTPSConnection("api.cloudflare.com", 443, timeout=20, context=ssl.create_default_context())
+ try:
+ connection.request("PUT", path, body=body, headers={"Authorization": "Bearer " + credential,
+ "Content-Type": "application/json", "Accept": "application/json"})
+ response = connection.getresponse()
+ if not 200 <= response.status <= 299:
+ raise PublisherError("Cloudflare KV write rejected")
+ raw = response.read(MAX_REPLY + 1)
+ if len(raw) > MAX_REPLY:
+ raise PublisherError("Cloudflare KV reply exceeds size bound")
+ result = parse_json(raw)
+ if not isinstance(result, dict) or result.get("success") is not True:
+ raise PublisherError("Cloudflare KV write not confirmed")
+ except (OSError, http.client.HTTPException):
+ raise PublisherError("Cloudflare KV write unavailable or unconfirmed") from None
+ finally:
+ connection.close()
+
+
+def ensure_directory(path):
+ try:
+ if not path.exists():
+ # 안전하지 않은 부모 디렉터리를 새로 만들지 않는다.
+ path.mkdir(mode=0o700)
+ if os.name == "nt":
+ windows_security(path, apply=True)
+ private_path(path, directory=True)
+ except OSError:
+ raise PublisherError("checkpoint directory unavailable") from None
+
+
+def create_private_file(path):
+ try:
+ fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600)
+ except FileExistsError:
+ private_path(path)
+ return os.open(path, os.O_RDWR | getattr(os, "O_NOFOLLOW", 0))
+ try:
+ if os.name == "nt":
+ windows_security(path, apply=True)
+ return fd
+ except BaseException:
+ os.close(fd)
+ raise
+
+
+@contextmanager
+def destination_lock(directory, identity):
+ path = directory / (identity + ".lock")
+ fd = create_private_file(path)
+ try:
+ with os.fdopen(fd, "r+b", buffering=0) as stream:
+ if os.name == "nt":
+ import msvcrt
+ if os.fstat(stream.fileno()).st_size == 0:
+ stream.write(b"\0")
+ stream.seek(0)
+ try:
+ msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1)
+ except OSError:
+ raise PublisherError("another publisher holds the destination lock") from None
+ else:
+ import fcntl
+ try:
+ fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+ except OSError:
+ raise PublisherError("another publisher holds the destination lock") from None
+ yield
+ except OSError:
+ raise PublisherError("checkpoint lock unavailable") from None
+
+
+def atomic_checkpoint(path, value):
+ temporary = None
+ try:
+ fd, temporary = tempfile.mkstemp(prefix=".snapshot-", dir=path.parent)
+ with os.fdopen(fd, "wb") as stream:
+ if os.name == "nt":
+ windows_security(Path(temporary), apply=True)
+ stream.write(encode_json(value))
+ stream.flush()
+ os.fsync(stream.fileno())
+ os.replace(temporary, path)
+ temporary = None
+ if os.name != "nt":
+ fd = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+ try:
+ os.fsync(fd)
+ finally:
+ os.close(fd)
+ except OSError:
+ raise PublisherError("atomic checkpoint failed; publication stopped") from None
+ finally:
+ if temporary is not None:
+ try:
+ os.unlink(temporary)
+ except OSError:
+ pass
+
+
+def utc_day(at):
+ return datetime.fromtimestamp(at / 1000, timezone.utc).date().isoformat()
+
+
+def load_checkpoint(path, target, at):
+ if not path.exists() and not path.is_symlink():
+ return {"version": 1, "destination": target, "day": utc_day(at), "attempts": 0,
+ "lastAttemptAt": None, "lastPublishedAt": None}
+ value = parse_json(private_read(path, MAX_REPLY))
+ if (not exact(value, "version destination day attempts lastAttemptAt lastPublishedAt")
+ or value["version"] != 1 or value["destination"] != target
+ or not isinstance(value["day"], str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value["day"])
+ or type(value["attempts"]) is not int or not 0 <= value["attempts"] <= DAILY_BUDGET
+ or not timestamp(value["lastAttemptAt"]) or not timestamp(value["lastPublishedAt"])):
+ raise PublisherError("checkpoint malformed or destination mismatch")
+ try:
+ datetime.strptime(value["day"], "%Y-%m-%d")
+ if (value["attempts"] > 0 and (value["lastAttemptAt"] is None
+ or utc_day(value["lastAttemptAt"]) != value["day"])):
+ raise ValueError("inconsistent reservation")
+ if (value["lastPublishedAt"] is not None and (value["lastAttemptAt"] is None
+ or value["lastPublishedAt"] > value["lastAttemptAt"])):
+ raise ValueError("inconsistent publication")
+ except (ValueError, OverflowError, OSError):
+ raise PublisherError("checkpoint malformed or inconsistent") from None
+ if value["day"] > utc_day(at) or (value["lastAttemptAt"] is not None and value["lastAttemptAt"] > at):
+ raise PublisherError("UTC clock moved backwards; refusing budget reset")
+ if value["day"] != utc_day(at):
+ value = {**value, "day": utc_day(at), "attempts": 0}
+ return value
+
+
+def reserve_checkpoint(value, at, interval):
+ if value["day"] != utc_day(at):
+ if value["day"] > utc_day(at):
+ raise PublisherError("UTC clock moved backwards")
+ value = {**value, "day": utc_day(at), "attempts": 0}
+ if value["attempts"] >= DAILY_BUDGET:
+ raise PublisherError("600-attempt UTC daily publication budget exhausted")
+ if value["lastAttemptAt"] is not None and at - value["lastAttemptAt"] < interval * 1000:
+ raise PublisherError("publication cadence guard; wait for the configured interval")
+ return {**value, "attempts": value["attempts"] + 1, "lastAttemptAt": at}
+
+
+def publish_once(args, fetch=fetch_lan, put=put_snapshot, clock=now_ms):
+ target = destination(args.account_id, args.namespace_id)
+ credential = read_credential(args, clock)
+ directory = Path(args.checkpoint_dir)
+ ensure_directory(directory)
+ identity = hashlib.sha256(target.encode("ascii")).hexdigest()
+ checkpoint = directory / (identity + ".json")
+ with destination_lock(directory, identity):
+ value = load_checkpoint(checkpoint, target, clock())
+ # LAN 접속 전 예산·간격을 확인하고 쓰기 예약 시 다시 대조한다.
+ reserve_checkpoint(value, clock(), args.interval)
+ snapshot, raw = collect_snapshot(fetch, clock)
+ reserved = reserve_checkpoint(value, clock(), args.interval)
+ atomic_checkpoint(checkpoint, reserved)
+ put(args.account_id, args.namespace_id, credential, raw)
+ confirmed = {**reserved, "lastPublishedAt": snapshot["publishedAt"]}
+ atomic_checkpoint(checkpoint, confirmed)
+ return {"published": True, "observedAt": snapshot["observedAt"], "publishedAt": snapshot["publishedAt"],
+ "bytes": len(raw), "utcAttempts": confirmed["attempts"]}
+
+
+def parser():
+ result = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+ modes = result.add_mutually_exclusive_group()
+ modes.add_argument("--plan", action="store_true", help="default: report fixed plan without I/O")
+ modes.add_argument("--validate", action="store_true", help="validate explicit destination and credential; no I/O to LAN/cloud")
+ modes.add_argument("--once", action="store_true", help="publish one snapshot to the existing namespace")
+ modes.add_argument("--loop", action="store_true", help="publish until SIGINT/SIGTERM; no service installation")
+ result.add_argument("--account-id")
+ result.add_argument("--namespace-id")
+ auth = result.add_mutually_exclusive_group()
+ auth.add_argument("--token-env", metavar="NAME", help="explicit environment variable containing machine KV token")
+ auth.add_argument("--token-file", help="private plaintext machine KV token file")
+ auth.add_argument("--operator-wrangler-auth-file", help="private existing Wrangler TOML session on operator PC only (broad OAuth)")
+ result.add_argument("--checkpoint-dir", help="persistent private directory; parent must exist")
+ result.add_argument("--interval", type=int, default=DEFAULT_INTERVAL, help="publication seconds, >=180; default 180 (~480/day)")
+ result.add_argument("--ssh-forward", action="store_true", help="operator source through an existing 127.0.0.1:8080 SSH forward to Pi5 localhost:8080")
+ return result
+
+
+def validate_arguments(args):
+ destination(args.account_id, args.namespace_id)
+ if args.interval < DEFAULT_INTERVAL:
+ raise PublisherError("publication interval must be at least 180 seconds")
+ if not any((args.token_env, args.token_file, args.operator_wrangler_auth_file)):
+ raise PublisherError("an explicit machine credential source or operator session is required")
+ if args.token_env and not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", args.token_env):
+ raise PublisherError("invalid credential environment variable name")
+ if not args.checkpoint_dir:
+ raise PublisherError("explicit persistent checkpoint directory is required")
+
+
+def main(argv=None):
+ args = parser().parse_args(argv)
+ if not (args.validate or args.once or args.loop):
+ print(json.dumps({"mode": "plan", "origin": f"http://{ORIGIN_HOST}:{ORIGIN_PORT}", "paths": sorted(READ_PATHS),
+ "key": SNAPSHOT_KEY, "intervalSeconds": args.interval, "utcAttemptBudget": DAILY_BUDGET,
+ "maxSnapshotBytes": MAX_SNAPSHOT, "resourcesCreated": False,
+ "required": "explicit existing account/namespace, credential source, persistent private checkpoint directory",
+ "quota": "account free quota is shared; zero billing is not guaranteed"}))
+ return 0
+ try:
+ validate_arguments(args)
+ if args.validate:
+ read_credential(args)
+ directory = Path(args.checkpoint_dir)
+ if directory.exists():
+ private_path(directory, directory=True)
+ else:
+ raise PublisherError("validation requires an existing private checkpoint directory")
+ print(json.dumps({"valid": True, "credentialMode": "operator-broad-oauth" if args.operator_wrangler_auth_file else "machine-kv-token", "networkContacted": False}))
+ return 0
+ stopped = threading.Event()
+ def stop(_signum, _frame):
+ stopped.set()
+ if args.loop:
+ signal.signal(signal.SIGINT, stop)
+ signal.signal(signal.SIGTERM, stop)
+ def fetch(path):
+ return fetch_lan(path, ssh_forward=args.ssh_forward)
+ while not stopped.is_set():
+ try:
+ print(json.dumps(publish_once(args, fetch=fetch)), flush=True)
+ except PublisherError as error:
+ print("snapshot publisher: " + str(error), file=sys.stderr, flush=True)
+ if not args.loop:
+ return 1
+ if not args.loop or stopped.wait(args.interval):
+ break
+ return 0
+ except PublisherError as error:
+ print("snapshot publisher: " + str(error), file=sys.stderr)
+ return 1
+ except (OSError, ValueError, TypeError, RecursionError, OverflowError):
+ # 예외 원문·자격증명·응답 본문·헤더를 로그에 남기지 않는다.
+ print("snapshot publisher: local configuration or I/O failure", file=sys.stderr)
+ return 1
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/apps/attendance/snapshot/test_publish.py b/apps/attendance/snapshot/test_publish.py
new file mode 100644
index 0000000..a01c46a
--- /dev/null
+++ b/apps/attendance/snapshot/test_publish.py
@@ -0,0 +1,444 @@
+"""Offline stdlib tests. Run explicitly with unittest; never contacts LAN/cloud."""
+import argparse
+import base64
+from copy import deepcopy
+from contextlib import redirect_stderr, redirect_stdout
+from datetime import datetime, timezone
+import hashlib
+import importlib.util
+import io
+import json
+import os
+from pathlib import Path
+import struct
+import tempfile
+import unittest
+from unittest.mock import patch
+import zlib
+
+spec = importlib.util.spec_from_file_location("attendance_snapshot_publish", Path(__file__).with_name("publish.py"))
+publisher = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(publisher)
+
+ACCOUNT = "a" * 32
+NAMESPACE = "b" * 32
+TOKEN = "machine-secret-do-not-disclose-1234567890"
+AT = 1_800_000_000_000
+
+
+def png(width=1, height=1):
+ def chunk(kind, data):
+ return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", zlib.crc32(kind + data))
+ return (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 0, 0, 0, 0))
+ + chunk(b"IDAT", zlib.compress((b"\0" + b"\xff" * width) * height)) + chunk(b"IEND", b""))
+
+
+def source_bodies():
+ return {
+ "/api/state": {"now": AT, "members": [{"name": "실제 회원", "present": True, "since": AT - 600_000,
+ "todayMs": 600_000, "weekMs": 1_200_000, "unrecordedWeek": 2,
+ "firstInToday": AT - 600_000, "lastCheckOut": None}], "unknownCards": [],
+ "recent": [{"name": "실제 회원", "checkIn": AT - 600_000, "checkOut": None, "unrecorded": False}]},
+ "/api/monitor": {"observedAt": AT - 10_000, "nodes": [{"id": identity, "label": identity,
+ "collectedAt": AT - 12_000, "cpuPercent": 23.75, "memoryUsedBytes": 123456,
+ "memoryTotalBytes": 654321, "health": "up"} for identity in ("prod", "stage", "db")]},
+ "/api/koin-health": {"observedAt": AT - 2000, "services": [{"id": identity, "label": identity,
+ "checkedAt": AT - 2500, "status": "up", "reason": "ready", "httpStatus": 200}
+ for identity in ("prod-api", "stage-api")]},
+ "/api/worker-health": {"checkedAt": AT - 5000, "workers": [{"id": "koin-web", "label": "WEB",
+ "status": "unknown", "httpStatus": 403, "reason": "restricted"}]},
+ "/api/spotify": {"state": "playing", "title": "Real track", "artists": ["Real artist"],
+ "updatedAt": AT - 7000, "coverUrl": "https://i.scdn.co/image/0123abc"},
+ "/api/jam": {"available": False},
+ }
+
+
+class FakeLAN:
+ def __init__(self):
+ self.bodies = source_bodies()
+ self.overrides = {}
+ self.calls = []
+
+ def __call__(self, path):
+ self.calls.append(path)
+ if path in self.overrides:
+ override = self.overrides[path]
+ if isinstance(override, Exception):
+ raise override
+ return override
+ if path in self.bodies:
+ return 200, "application/json", publisher.encode_json(self.bodies[path])
+ if path in publisher.ASSET_PATHS:
+ return 200, "image/png", png()
+ raise AssertionError("unexpected LAN endpoint")
+
+
+def arguments(directory):
+ return argparse.Namespace(account_id=ACCOUNT, namespace_id=NAMESPACE, token_env="SNAPSHOT_TEST_TOKEN",
+ token_file=None, operator_wrangler_auth_file=None,
+ checkpoint_dir=str(directory), interval=180)
+
+
+def make_private(path, directory=False):
+ if os.name == "nt":
+ publisher.windows_security(path, apply=True)
+ else:
+ path.chmod(0o700 if directory else 0o600)
+
+
+class SnapshotTests(unittest.TestCase):
+ def test_actual_data_and_timestamps_are_not_mutated(self):
+ lan = FakeLAN()
+ original = deepcopy(lan.bodies)
+ snapshot, raw = publisher.collect_snapshot(lan, lambda: AT + 8000)
+ self.assertEqual(lan.bodies, original)
+ self.assertEqual(snapshot["observedAt"], AT)
+ self.assertEqual(snapshot["publishedAt"], AT + 8000)
+ for path in publisher.JSON_PATHS:
+ self.assertEqual(snapshot["payloads"][path], {"status": 200, "body": original[path]})
+ self.assertEqual(json.loads(raw), snapshot)
+ self.assertEqual(base64.b64decode(snapshot["assets"]["/api/wifi-qr/5"]), png())
+ self.assertNotIn("/api/jam-qr", snapshot["assets"])
+ self.assertNotIn("/api/jam-qr", lan.calls)
+ self.assertEqual(set(lan.calls), set(publisher.JSON_PATHS + publisher.WIFI_PATHS))
+
+ def test_core_invalid_or_unavailable_aborts(self):
+ cases = [(503, "text/html", b"secret"), (200, "text/html", b"secret"),
+ (200, "application/json", b'{"now":1,"now":2}'),
+ (200, "application/json", b'{"now":NaN}'),
+ (200, "application/json", b"[]"), (200, "application/json", b"\xff")]
+ for response in cases:
+ with self.subTest(response=response[:2]):
+ lan = FakeLAN()
+ lan.overrides["/api/state"] = response
+ with self.assertRaises(publisher.PublisherError):
+ publisher.collect_snapshot(lan, lambda: AT)
+ for field in ("studentNumber", "queue", "deviceToken", "greeting"):
+ lan = FakeLAN()
+ lan.bodies["/api/state"][field] = "do-not-publish"
+ with self.assertRaises(publisher.PublisherError):
+ publisher.collect_snapshot(lan, lambda: AT)
+
+ def test_optional_error_status_preserved_but_secret_body_discarded(self):
+ lan = FakeLAN()
+ lan.overrides["/api/monitor"] = (401, "application/json", b'{"token":"do-not-publish"}')
+ lan.overrides["/api/koin-health"] = (503, "text/html", b"private diagnostics")
+ lan.overrides["/api/spotify"] = (302, "text/plain", b"https://secret.example")
+ lan.bodies["/api/worker-health"]["credentials"] = TOKEN
+ snapshot, raw = publisher.collect_snapshot(lan, lambda: AT)
+ self.assertEqual(snapshot["payloads"]["/api/monitor"]["status"], 401)
+ self.assertEqual(snapshot["payloads"]["/api/koin-health"]["status"], 503)
+ self.assertEqual(snapshot["payloads"]["/api/spotify"]["status"], 502)
+ self.assertEqual(snapshot["payloads"]["/api/worker-health"]["status"], 502)
+ for secret in ("do-not-publish", "private diagnostics", "secret.example", TOKEN):
+ self.assertNotIn(secret.encode(), raw)
+
+ def test_optional_malformed_json_or_untrusted_cover_is_unavailable(self):
+ for body in (b"{", b'{"state":"playing","state":"idle"}', b'{"state":Infinity}'):
+ lan = FakeLAN()
+ lan.overrides["/api/spotify"] = (200, "application/json", body)
+ snapshot, _ = publisher.collect_snapshot(lan, lambda: AT)
+ self.assertEqual(snapshot["payloads"]["/api/spotify"]["status"], 502)
+ for cover in ("http://i.scdn.co/image/abc", "https://i.scdn.co.evil/image/a", "https://user:password@i.scdn.co/image/a",
+ "https://i.scdn.co/image/a?token=secret", "https://192.168.1.112/camera"):
+ lan = FakeLAN()
+ lan.bodies["/api/spotify"]["coverUrl"] = cover
+ snapshot, raw = publisher.collect_snapshot(lan, lambda: AT)
+ self.assertEqual(snapshot["payloads"]["/api/spotify"]["status"], 502)
+ self.assertNotIn(cover.encode(), raw)
+
+ def test_wifi_absence_aborts_and_jam_is_never_invented(self):
+ for response in ((503, "application/json", b"{}"), (200, "image/png", b"fake PNG")):
+ lan = FakeLAN()
+ lan.overrides["/api/wifi-qr/2.4"] = response
+ with self.assertRaises(publisher.PublisherError):
+ publisher.collect_snapshot(lan, lambda: AT)
+ lan = FakeLAN()
+ lan.bodies["/api/jam"]["available"] = True
+ snapshot, _ = publisher.collect_snapshot(lan, lambda: AT)
+ self.assertIn("/api/jam-qr", snapshot["assets"])
+ lan.overrides["/api/jam-qr"] = (404, "application/json", b"{}")
+ snapshot, _ = publisher.collect_snapshot(lan, lambda: AT)
+ self.assertNotIn("/api/jam-qr", snapshot["assets"])
+ self.assertEqual(snapshot["payloads"]["/api/jam"]["status"], 502)
+
+ def test_png_bounds_crc_dimensions_and_stream(self):
+ self.assertTrue(publisher.valid_png(png()))
+ corrupt = bytearray(png())
+ corrupt[-1] ^= 1
+ for value in (b"", b"\x89PNG\r\n\x1a\n", png() + b"trailing", bytes(corrupt),
+ png(width=2049), png() + b"x" * publisher.MAX_PNG):
+ with self.subTest(length=len(value)):
+ self.assertFalse(publisher.valid_png(value))
+ def chunk(kind, data):
+ return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", zlib.crc32(kind + data))
+ header = png()[:33]
+ for decompressed in (b"\x05\0", b"\0" * 1_000_000):
+ value = header + chunk(b"IDAT", zlib.compress(decompressed)) + chunk(b"IEND", b"")
+ self.assertFalse(publisher.valid_png(value))
+
+ def test_snapshot_size_and_observation_guards(self):
+ lan = FakeLAN()
+ lan.bodies["/api/state"]["members"] *= 20_000
+ with self.assertRaisesRegex(publisher.PublisherError, "2 MiB"):
+ publisher.collect_snapshot(lan, lambda: AT)
+ for current in (AT - 1, AT + 60_001):
+ with self.assertRaises(publisher.PublisherError):
+ publisher.collect_snapshot(FakeLAN(), lambda: current)
+
+
+class HTTPTests(unittest.TestCase):
+ def test_fixed_targets_and_no_arbitrary_url_arguments(self):
+ for path in ("/api/camera", "/api/state?url=https://evil.example", "http://evil.example/", "//evil.example", "/api/queue"):
+ with patch.object(publisher.http.client, "HTTPConnection") as connection:
+ with self.assertRaises(publisher.PublisherError):
+ publisher.fetch_lan(path)
+ connection.assert_not_called()
+ for account, namespace in (("../evil", NAMESPACE), (ACCOUNT, "x" * 32), ("https://evil.example", NAMESPACE)):
+ with patch.object(publisher.http.client, "HTTPSConnection") as connection:
+ with self.assertRaises(publisher.PublisherError):
+ publisher.put_snapshot(account, namespace, TOKEN, b"{}")
+ connection.assert_not_called()
+
+ def test_lan_never_reads_errors_or_follows_redirects_or_sends_credentials(self):
+ with patch.object(publisher.http.client, "HTTPConnection") as connection:
+ response = connection.return_value.getresponse.return_value
+ response.status = 302
+ response.getheader.return_value = "text/html"
+ result = publisher.fetch_lan("/api/state")
+ self.assertEqual(result, (302, "text/html", b""))
+ response.read.assert_not_called()
+ connection.assert_called_once_with("192.168.1.112", 8080, timeout=12)
+ request = connection.return_value.request.call_args
+ self.assertEqual(request.args, ("GET", "/api/state"))
+ self.assertNotIn("Authorization", request.kwargs["headers"])
+ self.assertNotIn("Cookie", request.kwargs["headers"])
+
+ def test_https_put_is_single_fixed_nonexpiring_value(self):
+ with patch.object(publisher.http.client, "HTTPSConnection") as connection:
+ response = connection.return_value.getresponse.return_value
+ response.status = 200
+ response.read.return_value = b'{"success":true}'
+ publisher.put_snapshot(ACCOUNT, NAMESPACE, TOKEN, b"{}")
+ self.assertEqual(connection.call_args.args, ("api.cloudflare.com", 443))
+ request = connection.return_value.request.call_args
+ self.assertEqual(request.args, ("PUT", f"/client/v4/accounts/{ACCOUNT}/storage/kv/namespaces/{NAMESPACE}/values/clubroom:view:v1"))
+ self.assertEqual(request.kwargs["body"], b"{}")
+ self.assertEqual(request.kwargs["headers"]["Authorization"], "Bearer " + TOKEN)
+ self.assertNotIn("?", request.args[1])
+ self.assertEqual(connection.return_value.request.call_count, 1)
+
+ def test_cloud_error_body_never_read_or_leaked(self):
+ for status in (301, 302, 401, 429, 500):
+ with patch.object(publisher.http.client, "HTTPSConnection") as connection:
+ response = connection.return_value.getresponse.return_value
+ response.status = status
+ response.read.side_effect = AssertionError("error body must not be read")
+ with self.assertRaises(publisher.PublisherError) as error:
+ publisher.put_snapshot(ACCOUNT, NAMESPACE, TOKEN, b"{}")
+ self.assertNotIn(TOKEN, str(error.exception))
+ self.assertEqual(connection.return_value.request.call_count, 1)
+ for body in (b'{"success":false,"errors":["secret"]}', b"not-json", b"x" * (publisher.MAX_REPLY + 1)):
+ with patch.object(publisher.http.client, "HTTPSConnection") as connection:
+ response = connection.return_value.getresponse.return_value
+ response.status, response.read.return_value = 200, body
+ with self.assertRaises(publisher.PublisherError) as error:
+ publisher.put_snapshot(ACCOUNT, NAMESPACE, TOKEN, b"{}")
+ self.assertNotIn("secret", str(error.exception))
+
+
+class CredentialTests(unittest.TestCase):
+ def setUp(self):
+ self.temporary = tempfile.TemporaryDirectory()
+ self.directory = Path(self.temporary.name)
+ make_private(self.directory, directory=True)
+ self.args = arguments(self.directory)
+
+ def tearDown(self):
+ self.temporary.cleanup()
+
+ def credential_file(self, content):
+ path = self.directory / "credential"
+ path.write_text(content, encoding="utf-8")
+ make_private(path)
+ return path
+
+ def test_machine_environment_is_explicit_and_private_file_supported(self):
+ with patch.dict(os.environ, {"SNAPSHOT_TEST_TOKEN": TOKEN}, clear=True):
+ self.assertEqual(publisher.read_credential(self.args), TOKEN)
+ self.args.token_env = None
+ self.args.token_file = str(self.credential_file(TOKEN + "\n"))
+ self.assertEqual(publisher.read_credential(self.args), TOKEN)
+
+ def test_missing_invalid_or_public_credentials_fail_closed_without_leak(self):
+ for value in ("", "short", TOKEN + "\r\nInjected: header", "Bearer " + TOKEN):
+ with patch.dict(os.environ, {"SNAPSHOT_TEST_TOKEN": value}, clear=True):
+ with self.assertRaises(publisher.PublisherError) as error:
+ publisher.read_credential(self.args)
+ self.assertNotIn(TOKEN, str(error.exception))
+ if os.name != "nt":
+ path = self.credential_file(TOKEN)
+ path.chmod(0o644)
+ self.args.token_env, self.args.token_file = None, str(path)
+ with self.assertRaises(publisher.PublisherError):
+ publisher.read_credential(self.args)
+
+ def test_operator_oauth_is_explicit_private_expiry_checked_mode(self):
+ self.args.token_env = None
+ future = datetime.fromtimestamp((AT + 120_000) / 1000, timezone.utc).isoformat()
+ path = self.credential_file(f'oauth_token = "{TOKEN}"\nexpiration_time = "{future}"\n')
+ self.args.operator_wrangler_auth_file = str(path)
+ self.assertEqual(publisher.read_credential(self.args, lambda: AT), TOKEN)
+ with self.assertRaises(publisher.PublisherError):
+ publisher.read_credential(self.args, lambda: AT + 120_000)
+ path.write_text(f'oauth_token = "{TOKEN}"\nexpiration_time = "bad-expiry"\n', encoding="utf-8")
+ with self.assertRaises(publisher.PublisherError) as error:
+ publisher.read_credential(self.args, lambda: AT)
+ self.assertNotIn(TOKEN, str(error.exception))
+
+ def test_default_plan_performs_no_network_credential_or_checkpoint_io(self):
+ output = io.StringIO()
+ with patch.object(publisher, "read_credential", side_effect=AssertionError("credential I/O")), \
+ patch.object(publisher, "publish_once", side_effect=AssertionError("publish I/O")), \
+ redirect_stdout(output):
+ self.assertEqual(publisher.main([]), 0)
+ self.assertEqual(json.loads(output.getvalue())["mode"], "plan")
+
+ def test_cli_error_does_not_print_secret_or_raw_exception(self):
+ args = ["--once", "--account-id", ACCOUNT, "--namespace-id", NAMESPACE, "--token-env", "SNAPSHOT_TEST_TOKEN",
+ "--checkpoint-dir", str(self.directory)]
+ stderr, stdout = io.StringIO(), io.StringIO()
+ with patch.dict(os.environ, {"SNAPSHOT_TEST_TOKEN": TOKEN + "\nleak"}, clear=True), \
+ redirect_stderr(stderr), redirect_stdout(stdout), \
+ patch.object(publisher, "fetch_lan", side_effect=AssertionError("network must not be contacted")):
+ self.assertEqual(publisher.main(args), 1)
+ self.assertNotIn(TOKEN, stderr.getvalue() + stdout.getvalue())
+ with patch.object(publisher, "publish_once", side_effect=OSError(TOKEN)), \
+ redirect_stderr(stderr), redirect_stdout(stdout):
+ self.assertEqual(publisher.main(args), 1)
+ self.assertNotIn(TOKEN, stderr.getvalue() + stdout.getvalue())
+
+
+class CheckpointTests(unittest.TestCase):
+ def setUp(self):
+ self.temporary = tempfile.TemporaryDirectory()
+ self.directory = Path(self.temporary.name)
+ make_private(self.directory, directory=True)
+ self.args = arguments(self.directory)
+ self.target = publisher.destination(ACCOUNT, NAMESPACE)
+ self.identity = hashlib.sha256(self.target.encode("ascii")).hexdigest()
+ self.checkpoint = self.directory / (self.identity + ".json")
+ self.environment = patch.dict(os.environ, {"SNAPSHOT_TEST_TOKEN": TOKEN})
+ self.environment.start()
+
+ def tearDown(self):
+ self.environment.stop()
+ self.temporary.cleanup()
+
+ def load(self, at=AT):
+ return publisher.load_checkpoint(self.checkpoint, self.target, at)
+
+ def test_restart_keeps_counts_and_cadence_per_destination(self):
+ uploads = []
+ publisher.publish_once(self.args, FakeLAN(), lambda *args: uploads.append(args), lambda: AT)
+ self.assertEqual(self.load()["attempts"], 1)
+ self.assertEqual(self.load()["lastPublishedAt"], AT)
+ with self.assertRaisesRegex(publisher.PublisherError, "cadence"):
+ publisher.publish_once(self.args, FakeLAN(), lambda *args: uploads.append(args), lambda: AT + 1000)
+ self.assertEqual(len(uploads), 1)
+ lan = FakeLAN()
+ lan.bodies["/api/state"]["now"] = AT + 180_000
+ publisher.publish_once(self.args, lan, lambda *args: uploads.append(args), lambda: AT + 180_000)
+ self.assertEqual(self.load(AT + 180_000)["attempts"], 2)
+ other = deepcopy(self.args)
+ other.namespace_id = "c" * 32
+ publisher.publish_once(other, FakeLAN(), lambda *args: None, lambda: AT)
+ self.assertEqual(self.load(AT + 180_000)["attempts"], 2)
+ self.assertEqual(len(list(self.directory.glob("*.json"))), 2)
+
+ def test_utc_day_budget_guard_and_clock_rollback(self):
+ midnight = int(datetime(2026, 11, 13, tzinfo=timezone.utc).timestamp() * 1000)
+ value = publisher.load_checkpoint(self.checkpoint, self.target, midnight - 180_000)
+ value.update(attempts=600, lastAttemptAt=midnight - 180_000)
+ publisher.atomic_checkpoint(self.checkpoint, value)
+ with self.assertRaisesRegex(publisher.PublisherError, "budget"):
+ publisher.reserve_checkpoint(self.load(midnight - 1), midnight - 1, 180)
+ reset = self.load(midnight)
+ self.assertEqual(reset["attempts"], 0)
+ reserved = publisher.reserve_checkpoint(reset, midnight, 180)
+ self.assertEqual(reserved["attempts"], 1)
+ publisher.atomic_checkpoint(self.checkpoint, reserved)
+ with self.assertRaisesRegex(publisher.PublisherError, "backwards"):
+ self.load(midnight - 1)
+
+ def test_failed_cloud_write_preserves_success_checkpoint_but_consumes_attempt(self):
+ publisher.publish_once(self.args, FakeLAN(), lambda *args: None, lambda: AT)
+ later = AT + 180_000
+ lan = FakeLAN()
+ lan.bodies["/api/state"]["now"] = later
+ def fail(*args):
+ raise publisher.PublisherError("Cloudflare KV write unavailable or unconfirmed")
+ with self.assertRaises(publisher.PublisherError):
+ publisher.publish_once(self.args, lan, fail, lambda: later)
+ checkpoint = self.load(later)
+ self.assertEqual(checkpoint["attempts"], 2)
+ self.assertEqual(checkpoint["lastAttemptAt"], later)
+ self.assertEqual(checkpoint["lastPublishedAt"], AT)
+
+ def test_failed_reservation_stops_before_cloud_write(self):
+ put = unittest.mock.Mock()
+ with patch.object(publisher, "atomic_checkpoint", side_effect=publisher.PublisherError("atomic checkpoint failed")):
+ with self.assertRaises(publisher.PublisherError):
+ publisher.publish_once(self.args, FakeLAN(), put, lambda: AT)
+ put.assert_not_called()
+ self.assertFalse(self.checkpoint.exists())
+
+ def test_corrupt_or_cross_destination_checkpoint_never_resets_budget(self):
+ for raw in (b"{}", b"not-json", b'{"version":1,"version":1}', b'{"attempts":NaN}'):
+ self.checkpoint.write_bytes(raw)
+ make_private(self.checkpoint)
+ with self.assertRaises(publisher.PublisherError):
+ self.load()
+ value = {"version": 1, "destination": self.target + "evil", "day": publisher.utc_day(AT), "attempts": 10,
+ "lastAttemptAt": AT, "lastPublishedAt": AT}
+ publisher.atomic_checkpoint(self.checkpoint, value)
+ with self.assertRaises(publisher.PublisherError):
+ self.load()
+
+ def test_invalid_checkpoint_dates_and_reservations_fail_closed(self):
+ valid = {"version": 1, "destination": self.target, "day": publisher.utc_day(AT), "attempts": 1,
+ "lastAttemptAt": AT, "lastPublishedAt": AT}
+ for change in ({"day": "2026-99-99"}, {"lastAttemptAt": None}, {"lastPublishedAt": AT + 1},
+ {"day": "2020-01-01"}, {"attempts": True}):
+ publisher.atomic_checkpoint(self.checkpoint, {**valid, **change})
+ with self.assertRaises(publisher.PublisherError):
+ self.load()
+
+ def test_atomic_replace_failure_keeps_existing_checkpoint_and_removes_temporary(self):
+ value = publisher.load_checkpoint(self.checkpoint, self.target, AT)
+ publisher.atomic_checkpoint(self.checkpoint, value)
+ before = self.checkpoint.read_bytes()
+ with patch.object(publisher.os, "replace", side_effect=OSError("sensitive local detail")):
+ with self.assertRaisesRegex(publisher.PublisherError, "checkpoint failed"):
+ publisher.atomic_checkpoint(self.checkpoint, {**value, "attempts": 1})
+ self.assertEqual(self.checkpoint.read_bytes(), before)
+ self.assertEqual(list(self.directory.glob(".snapshot-*")), [])
+
+ def test_invalid_attendance_does_not_reserve_or_write(self):
+ lan = FakeLAN()
+ lan.overrides["/api/state"] = (503, "application/json", b"{}")
+ put = unittest.mock.Mock()
+ with self.assertRaises(publisher.PublisherError):
+ publisher.publish_once(self.args, lan, put, lambda: AT)
+ put.assert_not_called()
+ self.assertFalse(self.checkpoint.exists())
+
+ def test_lock_excludes_second_publisher(self):
+ with publisher.destination_lock(self.directory, self.identity):
+ with self.assertRaises(publisher.PublisherError):
+ with publisher.destination_lock(self.directory, self.identity):
+ self.fail("second publisher acquired lock")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/apps/attendance/src/dashboard.html b/apps/attendance/src/dashboard.html
deleted file mode 100644
index 51fcfa9..0000000
--- a/apps/attendance/src/dashboard.html
+++ /dev/null
@@ -1,4032 +0,0 @@
-
-
-
-
-
-
-
-
BCSD 출근부
-
-
-
-