mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 19:37:03 +01:00
Krzysztof kuhy Rudnicki
0c4b259c17
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""Registration and group endpoints.
|
|
|
|
Mirrors the shapes scripts/register_bot.sh depends on, so the whole runbook
|
|
can be rehearsed offline. The real captcha and SMS steps are the only parts
|
|
that cannot be faked, and both are represented by their exact failure mode.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
if TYPE_CHECKING:
|
|
from fake_signal.state import State
|
|
|
|
CAPTCHA_REQUIRED = (
|
|
"Captcha required for verification, use --captcha CAPTCHA\n"
|
|
"To get the token, go to https://signalcaptchas.org/registration/generate.html\n"
|
|
'After solving the captcha, right-click on the "Open Signal" link and copy the link.'
|
|
)
|
|
# Mirrors signalapp/registration-service in dev mode, which accepts the last
|
|
# six digits of the number as the verification code.
|
|
CODE_LENGTH = 6
|
|
|
|
|
|
def _state(request: Request) -> State:
|
|
"""Return the State attached to the running app."""
|
|
state: State = request.app.state.fake
|
|
return state
|
|
|
|
|
|
def expected_code(number: str) -> str:
|
|
"""The verification code the fake accepts for a number."""
|
|
digits = "".join(c for c in number if c.isdigit())
|
|
return digits[-CODE_LENGTH:]
|
|
|
|
|
|
def build_account_router() -> APIRouter:
|
|
"""Build the router covering accounts, registration and groups."""
|
|
router = APIRouter()
|
|
|
|
@router.get("/v1/accounts")
|
|
async def accounts(request: Request) -> list[str]:
|
|
return list(_state(request).accounts)
|
|
|
|
@router.post("/v1/register/{number}")
|
|
async def register(
|
|
request: Request, number: str, payload: dict[str, Any] | None = None
|
|
) -> JSONResponse:
|
|
captcha = (payload or {}).get("captcha")
|
|
if not captcha:
|
|
return JSONResponse({"error": CAPTCHA_REQUIRED}, status_code=400)
|
|
_state(request).pending[number] = expected_code(number)
|
|
return JSONResponse({}, status_code=201)
|
|
|
|
@router.post("/v1/register/{number}/verify/{token}")
|
|
async def verify(request: Request, number: str, token: str) -> JSONResponse:
|
|
state = _state(request)
|
|
if number not in state.pending:
|
|
return JSONResponse(
|
|
{"error": f"{number} has no registration in progress"}, status_code=400
|
|
)
|
|
if token != state.pending[number]:
|
|
return JSONResponse({"error": "invalid verification code"}, status_code=400)
|
|
del state.pending[number]
|
|
if number not in state.accounts:
|
|
state.accounts.append(number)
|
|
return JSONResponse({}, status_code=201)
|
|
|
|
return router
|