storage app

src/app/storage.py

@@ -0,0 +1,285 @@
+from fastapi import Response
+from .. import status
+from . import utils
+from fastapi.routing import APIRouter
+from pgsoft.pghash.md5 import md5
+
+
+router = APIRouter(
+    prefix="/storage",
+    tags=["storage"],
+)
+admin_token_md5 = "01112732d503cd3f0898d185a11c8b92"
+
+
+@router.post("/create_user")
+async def create_user(admin_token: str, email: str, response: Response) -> dict:
+    """ Create a normal user with user's email
+
+    Args:
+        admin_token (str): a token string of administrator
+        email (str): user's email
+
+    Returns:
+        dict: keys includes "status", "msg" and "data"
+    """
+    if md5(admin_token) != admin_token_md5:
+        response.status_code = status.HTTP_401_UNAUTHORIZED
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "Unauthorized"
+        }
+    email = email.lower().strip()
+    if utils.is_user_existed(email):
+        response.status_code = status.HTTP_304_NOT_MODIFIED
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "User already exists"
+        }
+    secret = utils.create_account(email, status.USER)
+    if secret is None:
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "creating normal user failed"
+        }
+    response.status_code = status.HTTP_201_CREATED
+    return {
+        "status": status.SUCCESS,
+        "data": {
+            "secret": secret,
+            "email": email
+        },
+        "msg": "created normal user successfully."
+    }
+
+
+@router.post("/register")
+async def register(secret: str, email: str, password: str, response: Response) -> dict:
+    """ Register an account with user's secret and password
+
+    Args:
+        secret (str): a random md5code provided by administrator
+        email (str): user's email
+        password (str): user's password
+
+    Returns:
+        dict: keys includes "status", "msg" and "data"
+    """
+    email = email.lower().strip()
+    if not utils.is_user_existed(email):
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "user not found"
+        }
+    if utils.is_user_registered(email):
+        response.status_code = status.HTTP_304_NOT_MODIFIED
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "you can't register repeatedly"
+        }
+    saved_secret = utils.get_secret(email)
+    if saved_secret is None:
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "secret not existed"
+        }
+    if saved_secret != secret:
+        response.status_code = status.HTTP_304_NOT_MODIFIED
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "incorrect secret or maybe your email is wrong",
+        }
+    if not utils.set_password(email, md5(password)):
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "failed to register"
+        }
+    if not utils.remove_user_secret(email):
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "failed to remove secret"
+        }
+    response.status_code = status.HTTP_200_OK
+    return {
+        "status": status.SUCCESS,
+        "data": {},
+        "msg": "Register Successfully, please login"
+    }
+
+
+@router.post("/login")
+async def login(email: str, password: str, response: Response) -> dict:
+    if email is None or email == "":
+        response.status_code = status.HTTP_400_BAD_REQUEST
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "email is empty"
+        }
+    if password is None or password == "":
+        response.status_code = status.HTTP_400_BAD_REQUEST
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "password is empty"
+        }
+    email = email.lower().strip()
+    if not utils.is_user_existed(email):
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "email does not exist"
+        }
+    password_digest = utils.get_password(email)
+    if password_digest is None:
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "password not existed"
+        }
+    password = md5(password)
+    if password_digest != password:
+        response.status_code = status.HTTP_400_BAD_REQUEST
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "incorrect password"
+        }
+    response.status_code = status.HTTP_200_OK
+    return {
+        "status": status.SUCCESS,
+        "data": {},
+        "msg": "login succeeded"
+    }
+
+
+@router.post("/delete_password")
+async def delete_password(admin_token: str, email: str, response: Response) -> dict:
+    """ delete password with user's email
+
+    Args:
+        admin_token (str): a token string of administrator
+        email (str): user's email
+
+    Returns:
+        dict: keys includes "status", "msg" and "data"
+    """
+    if md5(admin_token) != admin_token_md5:
+        response.status_code = status.HTTP_401_UNAUTHORIZED
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "Unauthorized"
+        }
+    email = email.lower().strip()
+    if not utils.is_user_existed(email):
+        response.status_code = status.HTTP_304_NOT_MODIFIED
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "User does not exists"
+        }
+    if not utils.get_password(email):
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "password not existed"
+        }
+    if not utils.delete_password(email):
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "failed to delete password"
+        }
+    secret = utils.generate_secret(email)
+    if secret is None:
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "creating secret failed"
+        }
+    response.status_code = status.HTTP_200_OK
+    return {
+        "status": status.SUCCESS,
+        "data": {
+            "secret": secret,
+            "email": email
+        },
+        "msg": "successfully deleted password."
+    }
+
+
+@router.post("/reset_password")
+async def reset_password(secret: str, email: str, password: str, response: Response) -> dict:
+    """ Reset password with user's email
+
+    Args:
+        secret (str): a random md5code provided by administrator
+        email (str): user's email
+        password (str): user's password
+
+    Returns:
+        dict: keys includes "status", "msg" and "data"
+    """
+    email = email.lower().strip()
+    if not utils.is_user_existed(email):
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "user not found"
+        }
+    saved_secret = utils.get_secret(email)
+    if saved_secret is None:
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "secret not existed"
+        }
+    if saved_secret != secret:
+        response.status_code = status.HTTP_304_NOT_MODIFIED
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "incorrect secret or maybe your email is wrong",
+        }
+    if not utils.set_password(email, md5(password)):
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "failed to change password"
+        }
+    if not utils.remove_user_secret(email):
+        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+        return {
+            "status": status.FAILURE,
+            "data": {},
+            "msg": "failed to remove secret"
+        }
+    response.status_code = status.HTTP_200_OK
+    return {
+        "status": status.SUCCESS,
+        "data": {},
+        "msg": "Reset password successfully, please login"
+    }

src/app/utils.py

@@ -0,0 +1,280 @@
+import huggingface_hub as hf
+import os
+from pgsoft import pgfile
+from pgsoft.pghash.md5 import md5
+from pgsoft.pgdate.date_utils import beijing
+
+
+hfapi = hf.HfApi()
+repo_id = "pgsoft/pguser"
+user_file_dir = "users"
+db_token = os.getenv("db_token")
+
+
+def file_exists(filename: str) -> bool:
+    """ Check if a file exist in dataset pguser
+
+    Args:
+        filename (str): relative path of the file
+
+    Returns:
+        bool: True if exist, False otherwise
+    """
+    return hfapi.file_exists(repo_id=repo_id,
+                             filename=filename,
+                             repo_type="dataset",
+                             token=db_token)
+
+
+def upload_file(localpath: str, remotepath: str, commit_message: str = None) -> bool:
+    """ upload a file to dataset pguser
+
+    Args:
+        localpath (str): the local path of the file
+        remotepath (str): the remote path of the file
+
+    Returns:
+        bool: True if success, False if failed
+    """
+    return pgfile.upload(localpath=localpath,
+                         remotepath=remotepath,
+                         repo_id=repo_id,
+                         repo_type="dataset",
+                         token=db_token,
+                         commit_message=commit_message)
+
+
+def download_file(remotepath: str) -> str | None:
+    """ download a file from dataset pguser
+
+    Args:
+        remotepath (str): the remote path of the file
+
+    Returns:
+        str | None: the local path of the file, 
+        return None if failed or file not exists
+    """
+    return pgfile.download(repo_id=repo_id,
+                           repo_type="dataset",
+                           remotepath=remotepath,
+                           token=db_token,
+                           localdir="./")
+
+
+def userdir(email: str) -> str:
+    """ the path to store userdata
+
+    Args:
+        email (str): user's email
+
+    Returns:
+        str: the path to store userdata
+    """
+    subdir = md5(email)[:5]
+    return f"{user_file_dir}/{subdir}/{email}"
+
+
+def is_user_existed(email: str) -> bool:
+    """ Check if user exists in dataset according to if user's role file exists
+
+    Args:
+        email (str): user's email
+
+    Returns:
+        bool: True if exists, False if not
+    """
+    filepath = f"{userdir(email)}/role"
+    return file_exists(filepath)
+
+
+def is_user_registered(email: str) -> bool:
+    """ Check if a existed user registered according to if user's password file exists
+
+    Args:
+        email (str): user's email
+
+    Returns:
+        bool: True if exists, False if not
+    """
+    filepath = f"{userdir(email)}/password"
+    return file_exists(filepath)
+
+
+def get_secret(email: str) -> str | None:
+    """ Get secret of user
+
+    Args:
+        email (str): user's email
+
+    Returns:
+        str: a random md5code of user saved in dataset
+    """
+    filepath = f"{userdir(email)}/secret"
+    localpath = download_file(filepath)
+    if localpath is None:
+        return None
+    try:
+        with open(localpath, "r") as f:
+            secret = f.read()
+        return secret
+    except Exception as e:
+        print(f"[get_secret]Failed to read secret, "
+              + f"{type(e)}: {e}")
+        return None
+
+
+def generate_user_secret() -> str:
+    """ Generate a unique random md5code as user's secret
+
+    Returns:
+        str: a random md5code
+    """
+    secret = md5(beijing())
+    return secret
+
+
+def remove_user_secret(email: str) -> bool:
+    """ remove user's secret
+
+    Args:
+        email (str): user's email
+
+    Returns:
+        bool: True if succeed, False if failed
+    """
+    filepath = f"{userdir(email)}/secret"
+    try:
+        hfapi.delete_file(repo_id=repo_id,
+                          repo_type="dataset",
+                          path_in_repo=filepath,
+                          token=db_token,
+                          commit_message="remove secret")
+        return True
+    except Exception as e:
+        print(f"[remove_user_secret]Failed to remove secret, "
+              + f"{type(e)}: {e}")
+        return False
+
+
+def create_account(email: str, role: str) -> str | None:
+    """ Create a account of designated role, which could be ADMIN or USER
+
+    Args:
+        email (str): user's email
+        role (str): the role of the account to be created
+
+    Returns:
+        str: the secret to be sent to user
+    """
+    prefix = userdir(email)
+    if not os.path.exists(prefix):
+        os.makedirs(prefix)
+    role_path = f"{prefix}/role"
+    secret_path = f"{prefix}/secret"
+    secret = generate_user_secret()
+    try:
+        with open(role_path, "w") as f:
+            f.write(role)
+        with open(secret_path, "w") as f:
+            f.write(secret)
+        if not upload_file(role_path, role_path, "create account"):
+            return None
+        if not upload_file(secret_path, secret_path, "create account"):
+            return None
+        return secret
+    except Exception as e:
+        print(f"[create_account]{type(e)}: {e}")
+        return None
+
+
+def set_password(email: str, password: str) -> bool:
+    """ save user's password to user's directory
+
+    Args:
+        email (str): user's email
+        password (str): password to save
+
+    Returns:
+        bool: True if succeed, False if failed
+    """
+    prefix = userdir(email)
+    if not os.path.exists(prefix):
+        os.makedirs(prefix)
+    filepath = f"{prefix}/password"
+    try:
+        with open(filepath, "w") as f:
+            f.write(password)
+        return upload_file(filepath, filepath, "set password")
+    except Exception as e:
+        print(f"[set_password] {type(e)}: {e}")
+        return False
+
+
+def get_password(email: str) -> str | None:
+    """ Get password of user
+
+    Args:
+        email (str): user's email
+
+    Returns:
+        str: password hashed with MD5
+    """
+    
+    filepath = f"{userdir(email)}/password"
+    localpath = download_file(filepath)
+    if localpath is None:
+        return None
+    try:
+        with open(localpath, "r") as f:
+            password = f.read()
+        return password
+    except Exception as e:
+        print(f"[get_password]Failed to read password, "
+              + f"{type(e)}: {e}")
+        return None
+
+
+def delete_password(email: str) -> bool:
+    """ delete user's password
+
+    Args:
+        email (str): user's email
+
+    Returns:
+        bool: True if succeed, False if failed
+    """
+    filepath = f"{userdir(email)}/password"
+    try:
+        hfapi.delete_file(repo_id=repo_id,
+                          repo_type="dataset",
+                          path_in_repo=filepath,
+                          token=db_token,
+                          commit_message="delete password")
+        return True
+    except Exception as e:
+        print(f"[delete_password]Failed to delete password, "
+              + f"{type(e)}: {e}")
+        return False
+
+
+def generate_secret(email: str) -> str | None:
+    """ Create a user secret
+    Args:
+        email (str): user's email
+    Returns:
+        str: the secret to be sent to user
+    """
+    prefix = userdir(email)
+    if not os.path.exists(prefix):
+        os.makedirs(prefix)
+    secret_path = f"{prefix}/secret"
+    secret = generate_user_secret()
+    try:
+        with open(secret_path, "w") as f:
+            f.write(secret)
+        if not upload_file(secret_path, secret_path, "create account"):
+            return None
+        return secret
+    except Exception as e:
+        print(f"[create_account]{type(e)}: {e}")
+        return None

src/main.py

@@ -1,15 +1,15 @@
 from fastapi import FastAPI
 import uvicorn
-from src.app import user
+from src.app import storage
 
 
 app = FastAPI()
-app.include_router(user.router)
+app.include_router(storage.router)
 
 
 @app.get("/")
 def root():
-    return {"detail": "PgSoft User API"}
+    return {"detail": "Storage server"}
 
 
 if __name__ == "__main__":

src/status.py

@@ -0,0 +1,10 @@
+from starlette.status import *
+
+
+SUCCESS = "OK"
+FAILURE = "Failure"
+
+# Roles
+ADMIN = "admin"
+USER = "user"
+GUEST = "guest"