(self)
| 2526 | self.send_error(404) |
| 2527 | |
| 2528 | def do_POST(self): |
| 2529 | p = urlparse(self.path).path.rstrip('/') |
| 2530 | length = int(self.headers.get('Content-Length', 0)) |
| 2531 | if length > MAX_REQUEST_BODY: |
| 2532 | self.send_json({'ok': False, 'error': f'Request body too large (max {MAX_REQUEST_BODY} bytes)'}, 413) |
| 2533 | return |
| 2534 | raw = self.rfile.read(length) if length else b'' |
| 2535 | try: |
| 2536 | body = json.loads(raw) if raw else {} |
| 2537 | except Exception: |
| 2538 | self.send_json({'ok': False, 'error': 'invalid JSON'}, 400) |
| 2539 | return |
| 2540 | |
| 2541 | # ── 认证端点（公开） ── |
| 2542 | if p == '/api/auth/setup': |
| 2543 | pw = body.get('password', '') |
| 2544 | if not isinstance(pw, str) or not pw: |
| 2545 | self.send_json({'ok': False, 'error': '请提供密码'}, 400) |
| 2546 | return |
| 2547 | self.send_json(setup_password(pw)) |
| 2548 | return |
| 2549 | if p == '/api/auth/login': |
| 2550 | pw = body.get('password', '') |
| 2551 | if not isinstance(pw, str) or not pw: |
| 2552 | self.send_json({'ok': False, 'error': '请提供密码'}, 400) |
| 2553 | return |
| 2554 | if verify_password(pw): |
| 2555 | token = create_token() |
| 2556 | resp = {'ok': True, 'token': token} |
| 2557 | # 同时设置 HttpOnly cookie |
| 2558 | try: |
| 2559 | body_bytes = json.dumps(resp, ensure_ascii=False).encode() |
| 2560 | self.send_response(200) |
| 2561 | self.send_header('Content-Type', 'application/json; charset=utf-8') |
| 2562 | self.send_header('Content-Length', str(len(body_bytes))) |
| 2563 | self.send_header('Set-Cookie', f'edict_token={token}; Path=/; HttpOnly; SameSite=Strict; Max-Age=86400') |
| 2564 | cors_headers(self) |
| 2565 | self.end_headers() |
| 2566 | self.wfile.write(body_bytes) |
| 2567 | except (BrokenPipeError, ConnectionResetError): |
| 2568 | pass |
| 2569 | else: |
| 2570 | self.send_json({'ok': False, 'error': '密码错误'}, 401) |
| 2571 | return |
| 2572 | |
| 2573 | # ── 认证检查 ── |
| 2574 | if self._check_auth(): |
| 2575 | return |
| 2576 | |
| 2577 | if p == '/api/morning-config': |
| 2578 | if not isinstance(body, dict): |
| 2579 | self.send_json({'ok': False, 'error': '请求体必须是 JSON 对象'}, 400) |
| 2580 | return |
| 2581 | allowed_keys = {'categories', 'keywords', 'custom_feeds', 'notification', 'feishu_webhook'} |
| 2582 | unknown = set(body.keys()) - allowed_keys |
| 2583 | if unknown: |
| 2584 | self.send_json({'ok': False, 'error': f'未知字段: {", ".join(unknown)}'}, 400) |
| 2585 | return |
nothing calls this directly
no test coverage detected