Use flask session store instead of DIYing
This commit is contained in:
parent
07fe8f6ffc
commit
9d1de005d0
4 changed files with 73 additions and 61 deletions
|
|
@ -26,31 +26,6 @@ def test_user_and_check_password(user):
|
|||
assert user.check_password('monkey')
|
||||
|
||||
|
||||
def test_jwt(user):
|
||||
user._id = (_id := ObjectId(randbytes(12)))
|
||||
token = user.jwt
|
||||
header, payload, sig = (base64_decode(part.replace('.', ''))
|
||||
for part in token.split('.'))
|
||||
header = json.loads(header)
|
||||
payload = json.loads(payload)
|
||||
assert header['alg'] == 'RS256'
|
||||
assert header['typ'] == 'JWT'
|
||||
assert set(header.keys()) == {'alg', 'typ'}
|
||||
# Note that JWT contents are visible to the user: this can be useful but
|
||||
# must be done with caution
|
||||
assert payload['email'] == user.email
|
||||
assert payload['name'] == user.name
|
||||
assert ObjectId(base64_decode(payload['_id'])) == user._id == _id
|
||||
assert set(payload.keys()) == {'email', 'name', '_id', 'admin', 'moderator'}
|
||||
|
||||
result = user.verify_jwt(token)
|
||||
assert result.email == user.email
|
||||
assert result.name == user.name
|
||||
assert result._id == user._id == _id
|
||||
assert not result.admin
|
||||
assert not result.moderator
|
||||
|
||||
|
||||
def test_store_and_retreive(user: User, database: Database):
|
||||
try:
|
||||
database.store_user(user)
|
||||
|
|
@ -73,13 +48,3 @@ def test_store_and_retreive_by_id(user: User, database: Database):
|
|||
finally:
|
||||
if id := user._id:
|
||||
database.delete_user(id)
|
||||
|
||||
def test_store_and_retreive_by_jwt(user: User, database: Database):
|
||||
try:
|
||||
token = database.store_user(user).jwt
|
||||
assert user._id is not None
|
||||
retreived = database.get_user_from_token(token)
|
||||
assert retreived == user
|
||||
finally:
|
||||
if id := user._id:
|
||||
database.delete_user(id)
|
||||
|
|
@ -2,7 +2,7 @@ from base64 import b64decode, b64encode
|
|||
from dataclasses import dataclass
|
||||
import json
|
||||
from random import randbytes
|
||||
from typing import Optional, Any
|
||||
from typing import Optional, Any, Self
|
||||
|
||||
from bson.objectid import ObjectId
|
||||
import scrypt
|
||||
|
|
@ -24,6 +24,12 @@ class JwtUser:
|
|||
moderator: bool
|
||||
admin: bool
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, data: dict) -> Self:
|
||||
_id = ObjectId(base64_decode(data.pop('_id')))
|
||||
return cls(_id=_id, **data)
|
||||
|
||||
|
||||
@dataclass
|
||||
class User:
|
||||
_id: Optional[ObjectId]
|
||||
|
|
@ -63,6 +69,11 @@ class User:
|
|||
|
||||
@property
|
||||
def public_fields(self):
|
||||
"""
|
||||
Session data is visible to client scripts.
|
||||
|
||||
This is a feature, not a bug; client scripts may need to gather login info.
|
||||
"""
|
||||
return {
|
||||
'_id': base64_encode(self._id.binary),
|
||||
"email": self.email,
|
||||
|
|
@ -73,18 +84,3 @@ class User:
|
|||
|
||||
def check_password(self, password: str) -> bool:
|
||||
return self.password_hash == scrypt.hash(password, self.salt)
|
||||
|
||||
@property
|
||||
def jwt(self) -> str:
|
||||
return jwt.encode(self.public_fields, PRIVATE_KEY, algorithm='RS256')
|
||||
|
||||
@staticmethod
|
||||
def verify_jwt(token: str) -> JwtUser:
|
||||
verified = jwt.decode(token, PUBLIC_KEY, verify=True, algorithms=['RS256'])
|
||||
return JwtUser(
|
||||
_id=ObjectId(base64_decode(verified['_id'])),
|
||||
name=verified['name'],
|
||||
email=verified['email'],
|
||||
moderator=verified['moderator'],
|
||||
admin=verified['admin'],
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue