auth_system.py

#auth_system.py
"""
Very basic user system — add username/password, authenticate.
"""

import os
import json
import bcrypt
import re

from logger import Logger, INFO, WARNING, ERROR, CRIT

class AuthSystem:
    def __init__(self, save_file="data/users.json"):
        self.users = {}
        self.save_file = save_file
        self.logger = Logger(True, "logs/auth_system_log.txt")

        self.load_users()



    def add_user(self, username, password, func_name="add_user"):
        self.logger.log_function_start(func_name)

        phases = [
            "Validating username",
            "Validating password",
            "Checking if duplicate username",
            "Hashing password",
            "Adding user",
            "Saving"
        ]

        phase_num = 0

        #Validate username
        phase_num = self.logger.advance_phase(phases, phase_num)

        username = username.strip()
        if not username:
            self.logger.log("Username may not be blank.", WARNING)
            self.logger.log_function_end(func_name, None, False)
            return False, "Invalid username"
        
        #Validate password
        phase_num = self.logger.advance_phase(phases, phase_num)

        if not self._validate_password(password):
            self.logger.log_function_end(func_name, None, False)
            return False, "Invalid password"
          
        #Check for duplicate username
        phase_num = self.logger.advance_phase(phases, phase_num)

        if self.user_exists(username):
            self.logger.log(f"Duplicate username attemped: {username}", ERROR)
            self.logger.log_function_end(func_name, None, False)
            return False, "Username already in use"
        
        #Hash password
        phase_num = self.logger.advance_phase(phases, phase_num)

        hashed = bcrypt.hashpw(password.encode(), bcrypt.gensalt())

        #Add user
        phase_num = self.logger.advance_phase(phases, phase_num)

        self.users[username] = hashed.decode()
        
        #Save
        phase_num = self.logger.advance_phase(phases, phase_num)

        self.save()
        self.logger.log_function_end(func_name, None)
        return True, None


    def delete_user(self, username, func_name="delete_user"):
        self.logger.log_function_start(func_name)

        phases = [
            "Searching for username.",
            "User located.",
            "Deleting user.",
            "Saving users"
        ]

        phase_num = 0

        phase_num = self.logger.advance_phase(phases, phase_num)
        
        #User found
        if username in self.users:
            phase_num = self.logger.advance_phase(phases, phase_num)
            
            #Delete user
            phase_num = self.logger.advance_phase(phases, phase_num)
            del self.users[username]
            self.logger.log(f"User {username} has been deleted.", INFO)
            
            #Save
            phase_num = self.logger.advance_phase(phases, phase_num)
            self.save()
            self.logger.log_function_end(func_name, None)
        
        #User not found
        else:
            self.logger.log(f"User {username} not found in users.", WARNING)
            self.logger.log_function_end(func_name, None, False)
    


    def change_password(self, username, old_password, new_password, func_name="change_password"):
        self.logger.log_function_start(func_name)

        phases = [
            "Authenticaing password",
            "Validating new password",
            "Hashing password",
            "Changing password",
            "Saving user"
        ]

        phase_num = 0

        #Authenticate old password
        phase_num = self.logger.advance_phase(phases, phase_num)
        
        if not self.authenticate(username, old_password):
            self.logger.log("Authentication failed", WARNING)

            #Incorrect password
            self.logger.log_function_end(func_name, None, False)
            return
        
        #Validate new password
        phase_num = self.logger.advance_phase(phases, phase_num)
        if not self._validate_password(new_password):
            self.logger.log_function_end(func_name, None, False)
            return
        
        
        #Hash new password
        phase_num = self.logger.advance_phase(phases, phase_num)
        self.users[username] = bcrypt.hashpw(new_password.encode(), bcrypt.gensalt()).decode()

        #Change password
        phase_num = self.logger.advance_phase(phases, phase_num)
        
        #Save
        phase_num = self.logger.advance_phase(phases, phase_num)
        self.save()

        #Exit
        self.logger.log_function_end(func_name, None)      
    


    def authenticate(self, username, password, func_name="authenticate"):
        self.logger.log_function_start(func_name)

        if username not in self.users:
            return False, "Incorrect Username"

        phases = [
            "Retrieving password",
            "Comparing passwords"
        ]

        phase_num = 0

        #Retrieve password
        phase_num = self.logger.advance_phase(phases, phase_num)
        hashed = self.users.get(username)
        
        if not hashed:
            #Password not found
            self.logger.log("Invalid user", CRIT)
            self.logger.log_function_end(func_name, False, False)
            return False, None
        
        #Compare passwords
        phase_num = self.logger.advance_phase(phases, phase_num)
        
        return_value = bcrypt.checkpw(password.encode(), hashed.encode())
        self.logger.log_function_end(func_name, return_value)
        return return_value, "Incorrect Password" if return_value == False else None



    def user_exists(self, username, func_name="user_exists"):
        self.logger.log_function_start(func_name)
        self.logger.log_function_end(func_name, username in self.users)
        return username in self.users
    


    def save(self, func_name="save"):
        self.logger.log_function_start(func_name)

        def create_required_diretories():
            for dir in required_dirs:
                if not os.path.exists(dir):
                    self.logger.log(f"Directories requirement not satisfied: [{dir}].", ERROR)
                    os.makedirs(dir)
                    self.logger.log(f"Directory created [{dir}].", INFO)
        
        required_dirs = ["data"]

        phases = [
            "Checking for required directories.",
            "Opening file",
            "Dumping contents"
        ]

        phase_num = 0

        phase_num = self.logger.advance_phase(phases, phase_num)
        
        #Create directories if don't exists
        create_required_diretories()
        
        #open file
        with open(self.save_file, 'w') as f:
            phase_num = self.logger.advance_phase(phases, phase_num)

            #Dump
            json.dump(self.users, f, indent=4, default=str)
            phase_num = self.logger.advance_phase(phases, phase_num)

        #Log
        self.logger.log("Save Successful.", INFO)
        self.logger.log_function_end(func_name, None)



    def load_users(self, func_name="load_users"):
        self.logger.log_function_start(func_name)

        phases = [
            "Checking path and file size.",
            "Opening File.",
            "Loading Content." 
        ]

        phase_num = 0

        #If save exists and is not empty
        phase_num = self.logger.advance_phase(phases, phase_num)

        if os.path.exists(self.save_file) and os.path.getsize(self.save_file) > 0:
            try:

                #Open file
                phase_num = self.logger.advance_phase(phases, phase_num)

                with open(self.save_file, 'r') as f:

                    #Users equal file contents
                    phase_num = self.logger.advance_phase(phases, phase_num)

                    self.users = json.load(f)
                    self.logger.log("Load Successful.", INFO)
                    self.logger.log_function_end(func_name, None)

            #Exceptions
            except (json.JSONDecodeError, IOError) as e:
                self.logger.log(f"Error loading users: {e}.", ERROR)
                self.logger.log("Defaulting to empty dictionary.", INFO)
                self.logger.log_function_end(func_name, None, False)
                self.users = {}
    


    def _validate_password(self, password, func_name = "_validate_password"):
        self.logger.log_function_start(func_name)

        warnings = [
            "Password must be at least 8 characters long.",
            "Password must contain at least one uppercase letter.",
            "Password must contain at least one lowercase letter.",
            "Password must contain at least one digit.",
            "Password must contain at least one symbol."
        ]

        phases = [
            "Checking password is required length.",
            "Checking password has required characters"
        ]
        phase_num = 0

        error_indices = []
        checks = True

        
        phase_num = self.logger.advance_phase(phases, phase_num)

        #Length greater than 8 characters
        if len(password) < 8:
            error_indices.append(0)
            checks = False
        
        phase_num = self.logger.advance_phase(phases, phase_num)
        
        #Must have at least one uppercase 
        if not re.search(r"[A-Z]", password):
            error_indices.append(1)
            checks = False
        
        #Must have at least one lowercase
        if not re.search(r"[a-z]", password):
            error_indices.append(2)
            checks = False
        
        #Must have at least one digit
        if not re.search(r"\d", password):
            error_indices.append(3)
            checks = False
        
        #Must have at least one symbol
        if not re.search(r"[^\w\s]", password):
            error_indices.append(4)
            checks = False
        
        #Invalid password
        if not checks:

            for index in error_indices:
                self.logger.log(warnings[index], WARNING)

            self.logger.log_function_end(func_name, False, False)
            return False
        
        #Valid password
        self.logger.log_function_end(func_name, True)
        return True