168 lines
4.7 KiB
Python
168 lines
4.7 KiB
Python
import json
|
|
|
|
from fastapi import Depends, FastAPI, HTTPException, status
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
|
|
from auth import Hasher, create_access_token, get_current_user
|
|
from crypto import (
|
|
deserialize_into_bytes,
|
|
fernet_decrypt,
|
|
fernet_encrypt,
|
|
generate_random_encryption_key,
|
|
generate_user_passkey,
|
|
serialize_bytes,
|
|
)
|
|
from models import Secret, User, UserLogin
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
origins = [
|
|
'http://localhost',
|
|
'http://localhost:5173',
|
|
"*"
|
|
]
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=['*'],
|
|
allow_headers=['*']
|
|
)
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"'message'": "Hello World"}
|
|
|
|
|
|
@app.post('/register')
|
|
async def register(user: User):
|
|
"""Registers a user"""
|
|
|
|
users = []
|
|
with open('database/users.json', 'r') as f:
|
|
text = f.read()
|
|
if text:
|
|
users = json.loads(text)
|
|
|
|
if user.id is not None:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="User id shall be auto generated, cannot be provided in request"
|
|
)
|
|
if not users:
|
|
user.id = 0
|
|
else:
|
|
max_user_id = max([i['id'] for i in users])
|
|
user.id = max_user_id + 1
|
|
|
|
user_exists = [i for i in users if i['username'] == user.username]
|
|
if user_exists:
|
|
raise HTTPException(status_code=400, detail="Username already in use")
|
|
|
|
encryption_key = generate_random_encryption_key()
|
|
|
|
salt, master_key = generate_user_passkey(user.password)
|
|
encrypted_encryption_key = fernet_encrypt(encryption_key, master_key)
|
|
|
|
user.password = Hasher.get_password_hash(user.password)
|
|
user.encryption_key = encrypted_encryption_key.decode('utf-8')
|
|
user.salt = serialize_bytes(salt)
|
|
|
|
users.append(jsonable_encoder(user))
|
|
# print(f"{salt=}\n{user.salt=}\n{encrypted_encryption_key=}\n{user.encryption_key=}\n{master_key=}")
|
|
with open('database/users.json', 'w') as f:
|
|
json.dump(users, f)
|
|
|
|
return {'user_id': user.id}
|
|
|
|
|
|
@app.post('/login')
|
|
async def login(user: UserLogin):
|
|
"""logs in the user"""
|
|
|
|
users = []
|
|
with open('database/users.json', 'r') as f:
|
|
text = f.read()
|
|
if text:
|
|
users.extend(json.loads(text))
|
|
|
|
cur_user = [i for i in users if i['username']==user.username]
|
|
if not cur_user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="username or password is incorrect"
|
|
)
|
|
else:
|
|
cur_user = cur_user[0]
|
|
|
|
password_match = Hasher.verify_password(user.password, cur_user['password'])
|
|
if not password_match:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="username or password is incorrect"
|
|
)
|
|
|
|
encrypted_encryption_key = cur_user['encryption_key'].encode()
|
|
salt = deserialize_into_bytes(cur_user['salt'])
|
|
_, master_key = generate_user_passkey(user.password, salt)
|
|
encryption_key = fernet_decrypt(encrypted_encryption_key, master_key)
|
|
access_token = create_access_token(subject=cur_user['id'], encryption_key=encryption_key)
|
|
|
|
response = {
|
|
'message': 'authenticated',
|
|
'accessToken': access_token
|
|
}
|
|
|
|
return response
|
|
|
|
|
|
@app.post("/secret")
|
|
async def create_secret(secret: Secret, current_user: dict = Depends(get_current_user)):
|
|
"""
|
|
Stores and encrypted secret for the user.
|
|
|
|
The encrypted secret is unreadable on the server and is encrypted on the front-end
|
|
"""
|
|
|
|
data = []
|
|
with open('database/secrets.json', 'r') as f:
|
|
text = f.read()
|
|
if text:
|
|
data.extend(json.loads(text))
|
|
secret.user_id = current_user['id']
|
|
encryption_key = current_user['encryption_key'].encode()
|
|
|
|
encrypted_data = fernet_encrypt(secret.data.encode(), encryption_key)
|
|
|
|
secret.data = encrypted_data.decode('utf-8')
|
|
data.append(jsonable_encoder(secret))
|
|
|
|
with open('database/secrets.json', 'w') as f:
|
|
json.dump(data, f)
|
|
return secret
|
|
|
|
|
|
@app.get('/secret')
|
|
async def list_secret(current_user: dict = Depends(get_current_user)):
|
|
"""Returns the encrypted secrets of the user."""
|
|
|
|
data = []
|
|
with open('database/secrets.json', 'r') as f:
|
|
text = f.read()
|
|
if text:
|
|
data.extend(json.loads(text))
|
|
|
|
user_id = current_user['id']
|
|
encryption_key = current_user['encryption_key'].encode()
|
|
|
|
user_secrets = [i for i in data if i['user_id']==user_id]
|
|
for secret in user_secrets:
|
|
cur_data = secret['data']
|
|
decrypted_data = fernet_decrypt(cur_data, encryption_key)
|
|
secret['data'] = decrypted_data
|
|
|
|
return user_secrets |