auth.py

@@ -9,8 +9,6 @@ from fastapi import Depends, HTTPException, status
 from fastapi.security import HTTPBearer
 from jwt.exceptions import InvalidTokenError
 
-import queries
-
 ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 30 minutes
 ALGORITHM = "HS256"
 JWT_SECRET_KEY = 'abcdefghijklmnopqrstuvwxyz'
@@ -49,14 +47,19 @@ def get_current_user(token: str = Depends(security)) -> dict:
     except InvalidTokenError:
         raise credentials_exception
 
+    with open('database/users.json', 'r') as f:
+        text = f.read()
+        if text:
+            data = json.loads(text)
+        else:
+            raise credentials_exception
 
-    user = queries.GET_USER_BY_ID({'user_id': user_id})
-
-    if user is None:
+    user = [i for i in data if i['id']==user_id]
+    if not user:
         raise credentials_exception
 
     cur_user = {'id': user_id}
-    cur_user['username'] = user['username']
+    cur_user['username'] = user[0]['username']
     cur_user['encryption_key'] = payload['key']
 
     return cur_user

main.py

@@ -1,17 +1,20 @@
 import json
-from sqlite3 import IntegrityError
-from typing import Literal
 
 from fastapi import Depends, FastAPI, HTTPException, status
 from fastapi.encoders import jsonable_encoder
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.security import OAuth2PasswordBearer
 
-import queries
 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 Key, Secret, User, UserLogin
+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()
 
@@ -39,51 +42,72 @@ async def root():
 async def register(user: User):
     """Registers a user"""
 
-    plain_text_password = user.password
-    user.password = Hasher.get_password_hash(plain_text_password)
+    users = []
+    with open('database/users.json', 'r') as f:
+        text = f.read()
+        if text:
+            users = json.loads(text)
 
-    try:
-        user = queries.CREATE_USER(user.model_dump())
-        user_id = user['id']
-    except IntegrityError as e:
+    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(plain_text_password)
 
-    key = Key(
-        user_id=user_id,
-        encryption_key=fernet_encrypt(encryption_key, master_key).decode('utf-8'),
-        encryption_key_salt=serialize_bytes(salt)
-    )
+    salt, master_key = generate_user_passkey(user.password)
+    encrypted_encryption_key = fernet_encrypt(encryption_key, master_key)
 
-    try:
-        queries.CREATE_KEY(key.model_dump())
-    except Exception as e:
-        print('failed to create key', e)
+    user.password = Hasher.get_password_hash(user.password)
+    user.encryption_key = encrypted_encryption_key.decode('utf-8')
+    user.salt = serialize_bytes(salt)
 
-    return {'user_id': user_id}
+    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"""
 
-    login_exception = HTTPException(
-        status_code=status.HTTP_400_BAD_REQUEST,
-        detail="username or password is incorrect"
-    )
+    users = []
+    with open('database/users.json', 'r') as f:
+        text = f.read()
+        if text:
+            users.extend(json.loads(text))
 
-    cur_user = queries.GET_USER_WITH_KEY(user.model_dump())
-    if cur_user is None:
-        raise login_exception
+    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 login_exception
+        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['encryption_key_salt'])
+    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)
@@ -102,15 +126,27 @@ async def create_secret(secret: Secret, current_user: dict = Depends(get_current
     Stores an encrypted secret for the user.
     """
 
+    data = []
+    with open('database/secrets.json', 'r') as f:
+        text = f.read()
+        if text:
+            data.extend(json.loads(text))
+
+    if data:
+        secret_id = max(i['id'] for i in data) + 1
+    else:
+        secret_id = 0
+    secret.id = secret_id
     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))
 
-    queries.CREATE_SECRET(secret.model_dump())
-
+    with open('database/secrets.json', 'w') as f:
+        json.dump(data, f)
     return secret
 
 
@@ -152,9 +188,16 @@ async def update_secret(secret: Secret, current_user: dict = Depends(get_current
 async def list_secret(current_user: dict = Depends(get_current_user)):
     """Returns the encrypted secrets of the user."""
 
-    user_secrets = queries.GET_SECRETS({'user_id': current_user['id']})
+    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 and i['active']]
     for secret in user_secrets:
         cur_data = secret['data']
         decrypted_data = fernet_decrypt(cur_data, encryption_key)
@@ -166,7 +209,8 @@ async def list_secret(current_user: dict = Depends(get_current_user)):
 @app.get('/validate-token')
 async def validate_token(current_user: dict = Depends(get_current_user)):
     user_id = current_user['id']
-    if user_id is None:
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+    print("user_id: ", user_id)
+    if user_id is not None:
+        return {'message': 'authenticated'}
 
-    return {'message': 'authenticated'}
+    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

migration.py

@@ -5,9 +5,7 @@ import sqlite3
 
 
 def connect_db():
-    conn = sqlite3.connect('database/database.sqlite')
-    conn.execute("PRAGMA foreign_keys = 1")
-    return conn
+    return sqlite3.connect('database/database.sqlite')
 
 
 def create_user_table():
@@ -37,10 +35,10 @@ def create_user_table():
 def create_keys_table():
     query = """
     create table if not exists keys (
-        user_id                 INTEGER,
-        encryption_key          TEXT,
-        encryption_key_salt     TEXT,
-        FOREIGN KEY (user_id)   REFERENCES users (id)
+        user_id INTEGER,
+        encryption_key TEXT,
+        encryption_key_salt TEXT,
+        FOREIGN KEY (user_id) REFERENCES users (id)
         ON DELETE CASCADE
     )
     """
@@ -58,13 +56,13 @@ def create_keys_table():
 def create_secrets_table():
     query = """
     create table if not exists secrets (
-        id          INTEGER     PRIMARY KEY     AUTOINCREMENT,
-        user_id     INTEGER,
-        data        TEXT        NOT NULL,
-        added_on    TEXT        DEFAULT CURRENT_TIMESTAMP,
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        user_id INTEGER,
+        data TEXT NOT NULL,
+        added_on TEXT DEFAULT CURRENT_TIMESTAMP,
         modified_on TEXT,
-        active      BOOLEAN     DEFAULT TRUE,
-        FOREIGN KEY (user_id)   REFERENCES users(id)
+        active BOOLEAN DEFAULT TRUE,
+        FOREIGN KEY (user_id) REFERENCES users(id)
         ON DELETE CASCADE
     )
     """

models.py

@@ -5,20 +5,13 @@ from pydantic import BaseModel
 
 class User(BaseModel):
     id: int = None
-    name: str = None
-    email: str = None
     username: str
     password: str
+    encryption_key: str = None
+    salt: str = None
     created_on: str = datetime.datetime.now(datetime.timezone.utc).isoformat()
     active: bool = True
 
-
-class Key(BaseModel):
-    user_id: int
-    encryption_key: str
-    encryption_key_salt: str
-
-
 class UserLogin(BaseModel):
     username: str
     password: str

queries.py

@@ -1,140 +0,0 @@
-import sqlite3
-from dataclasses import dataclass
-from typing import Literal
-
-
-def connect_db():
-    conn = sqlite3.connect('database/database.sqlite')
-    conn.execute("PRAGMA foreign_keys = 1")
-    return conn
-
-
-class DbQuery:
-    def __init__(
-        self,
-        query: str,
-        type: Literal['read', 'write'],
-        returns: bool,
-        rows: Literal['single', 'multi']
-    ):
-        self.query = query
-        self.type = type
-        self.returns = returns
-        self.rows = rows
-
-    def __str__(self) -> str:
-        return self.query
-
-    def __repr__(self) -> str:
-        return f"<query: '{self.query}, {self.type}, returns {self.returns}, {self.rows} row(s)"
-
-    def _read(self, params: dict) -> list[dict]|dict:
-        with connect_db() as conn:
-            cur = conn.execute(self.query, params)
-            data = cur.fetchall()
-            if not data:
-                return None
-            keys = [i[0] for i in cur.description]
-
-        data_dict = [dict(zip(keys, i)) for i in data]
-
-        if self.rows == 'single':
-            if len(data_dict) > 1:
-                raise ValueError('Query returned more than one rows.')
-
-            data_dict = data_dict[0]
-
-        return data_dict
-
-    def _write(self, params: dict) -> dict|bool:
-        if not isinstance(params, dict):
-            raise TypeError("Only dict type is supported for params")
-
-        with connect_db() as conn:
-            cur = conn.execute(self.query, params)
-            if self.returns:
-                data = cur.fetchone()
-                keys = keys = [i[0] for i in cur.description]
-                data_dict = dict(zip(keys, data))
-            conn.commit()
-
-        if self.returns:
-            return data_dict
-        return True
-
-    def __call__(self, params):
-        if self.type == 'read':
-            func = self._read
-        else:
-            func = self._write
-
-        return func(params)
-
-
-CREATE_USER = DbQuery(
-    query="""
-        INSERT INTO
-        users (name, email, username, password)
-        VALUES (:name, :email, :username, :password)
-        RETURNING id
-    """,
-    type='write',
-    returns=True,
-    rows='single'
-)
-
-CREATE_KEY = DbQuery(
-    query="""
-        INSERT INTO
-        keys (user_id, encryption_key, encryption_key_salt)
-        VALUES (:user_id, :encryption_key, :encryption_key_salt)
-    """,
-    type='write',
-    returns=False,
-    rows=None
-)
-
-GET_USER_WITH_KEY = DbQuery(
-    """
-        SELECT *
-        FROM users u
-        JOIN keys k on u.id = k.user_id
-        WHERE u.username = :username
-    """,
-    type='read',
-    returns=True,
-    rows='single'
-)
-
-GET_USER_BY_ID = DbQuery(
-    """
-        SELECT *
-        FROM users
-        where id = :user_id
-    """,
-    type='read',
-    returns=True,
-    rows='single'
-)
-
-CREATE_SECRET = DbQuery(
-    """
-        INSERT INTO
-        secrets(user_id, data)
-        VALUES(:user_id, :data)
-    """,
-    type='write',
-    returns=False,
-    rows=None
-)
-
-GET_SECRETS = DbQuery(
-    """
-        SELECT *
-        FROM secrets
-        WHERE user_id = :user_id
-    """,
-    type='read',
-    returns=True,
-    rows=None
-)