105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
"""전화번호 확인(문자 인증). 문자 발송 방법은 SmsSender 하나만 바꾸면 된다 (업체에 묶이지 않는다).
|
|
|
|
- 6자리 번호를 문자로 보내고, 5분 안에 맞게 넣으면 그 기기의 토큰이 번호에 묶인다.
|
|
- 번호당 1시간에 3번까지만 보낼 수 있고, 번호 하나에 5번 틀리면 그 번호는 새로 요청해야 한다.
|
|
- 서버에는 번호 자체가 아니라 소금 친 해시만 남긴다.
|
|
"""
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import secrets
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
|
|
|
|
class SmsSender:
|
|
"""문자 보내기 약속. send 가 True 를 돌려주면 보낸 것으로 본다."""
|
|
|
|
def send(self, phone: str, text: str) -> bool:
|
|
raise NotImplementedError
|
|
|
|
|
|
class LogSmsSender(SmsSender):
|
|
"""개발·시험용: 문자를 보내지 않고 파일(없으면 표준 출력)에 적는다. 실제 서비스에서 쓰지 말 것."""
|
|
|
|
def __init__(self, path=None):
|
|
self.path = path
|
|
|
|
def send(self, phone, text):
|
|
line = f"{phone}\t{text}\n"
|
|
if self.path:
|
|
with open(self.path, "a", encoding="utf-8") as f:
|
|
f.write(line)
|
|
else:
|
|
print("SMS", line, end="", flush=True)
|
|
return True
|
|
|
|
|
|
class WebhookSmsSender(SmsSender):
|
|
"""업체 중립: 주소 하나로 {"to":..., "text":...} 를 POST 한다. 업체별 변환은 그 주소 뒤에서 한다."""
|
|
|
|
def __init__(self, url, bearer=None, timeout=8):
|
|
self.url, self.bearer, self.timeout = url, bearer, timeout
|
|
|
|
def send(self, phone, text):
|
|
req = urllib.request.Request(
|
|
self.url, method="POST", data=json.dumps({"to": phone, "text": text}).encode("utf-8"),
|
|
headers={"Content-Type": "application/json", **({"Authorization": f"Bearer {self.bearer}"} if self.bearer else {})},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=self.timeout) as r:
|
|
return 200 <= r.status < 300
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
class Verifier:
|
|
CODE_TTL = 300
|
|
MAX_SENDS = 3
|
|
SEND_WINDOW = 3600
|
|
MAX_TRIES = 5
|
|
|
|
def __init__(self, sender: SmsSender, clock=time.time):
|
|
self._sender, self._clock = sender, clock
|
|
self._lock = threading.Lock()
|
|
self._pending = {} # phone -> [salt, hash, expires, tries]
|
|
self._sends = {} # phone -> [시각...]
|
|
|
|
@staticmethod
|
|
def _hash(salt: str, code: str) -> str:
|
|
return hashlib.sha256((salt + code).encode()).hexdigest()
|
|
|
|
def start(self, phone: str) -> str:
|
|
"""'ok' | 'rate' | 'fail'"""
|
|
now = self._clock()
|
|
with self._lock:
|
|
sent = [t for t in self._sends.get(phone, []) if now - t < self.SEND_WINDOW]
|
|
if len(sent) >= self.MAX_SENDS:
|
|
self._sends[phone] = sent
|
|
return "rate"
|
|
code = f"{secrets.randbelow(1_000_000):06d}"
|
|
salt = secrets.token_hex(8)
|
|
self._pending[phone] = [salt, self._hash(salt, code), now + self.CODE_TTL, 0]
|
|
self._sends[phone] = sent + [now]
|
|
ok = self._sender.send(phone, f"[Friendly Map] 확인 번호 {code} (5분 안에 입력)")
|
|
if not ok:
|
|
with self._lock:
|
|
self._pending.pop(phone, None)
|
|
self._sends[phone] = self._sends[phone][:-1]
|
|
return "fail"
|
|
return "ok"
|
|
|
|
def confirm(self, phone: str, code: str) -> bool:
|
|
now = self._clock()
|
|
with self._lock:
|
|
p = self._pending.get(phone)
|
|
if p is None or now > p[2] or p[3] >= self.MAX_TRIES:
|
|
self._pending.pop(phone, None)
|
|
return False
|
|
p[3] += 1
|
|
if hmac.compare_digest(p[1], self._hash(p[0], code.strip())):
|
|
self._pending.pop(phone, None)
|
|
return True
|
|
return False
|