46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
import hashlib
|
|
import hmac
|
|
import secrets
|
|
import threading
|
|
import time
|
|
from fastapi import HTTPException
|
|
|
|
|
|
def digest(value):
|
|
return hashlib.sha256(value.encode()).hexdigest()
|
|
|
|
|
|
def hash_password(password):
|
|
salt=secrets.token_bytes(16)
|
|
key=hashlib.scrypt(password.encode(),salt=salt,n=16384,r=8,p=1,dklen=32)
|
|
return f'scrypt${salt.hex()}${key.hex()}'
|
|
|
|
|
|
def verify_password(password,encoded):
|
|
try:
|
|
algorithm,salt,key=encoded.split('$')
|
|
if algorithm != 'scrypt':
|
|
return False
|
|
actual=hashlib.scrypt(password.encode(),salt=bytes.fromhex(salt),n=16384,r=8,p=1,dklen=32)
|
|
return hmac.compare_digest(actual,bytes.fromhex(key))
|
|
except (ValueError,TypeError):
|
|
return False
|
|
|
|
|
|
DUMMY_HASH=hash_password('never-a-real-account-password')
|
|
|
|
|
|
class RateLimit:
|
|
def __init__(self):
|
|
self.entries={}
|
|
self.lock=threading.Lock()
|
|
|
|
def check(self,key,limit=20,window=60):
|
|
now=time.monotonic()
|
|
with self.lock:
|
|
self.entries={k:v for k,v in self.entries.items() if now < v[0]}
|
|
expiry,count=self.entries.get(key,(now+window,0))
|
|
if count >= limit:
|
|
raise HTTPException(429,'操作过于频繁，请稍后重试')
|
|
self.entries[key]=(expiry,count+1)
|