Create auth.py

modules/auth.py

@@ -0,0 +1,104 @@
+from azure.cosmos import CosmosClient, exceptions
+import bcrypt
+import os
+
+# Azure Cosmos DB configuration
+endpoint = os.environ.get("COSMOS_ENDPOINT")
+key = os.environ.get("COSMOS_KEY")
+database_name = "user_database"
+container_name = "users"
+
+# Initialize the Cosmos client
+client = CosmosClient(endpoint, key)
+database = client.get_database_client(database_name)
+container = database.get_container_client(container_name)
+
+def hash_password(password):
+    """Hash a password for storing."""
+    return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
+
+def verify_password(stored_password, provided_password):
+    """Verify a stored password against one provided by user"""
+    return bcrypt.checkpw(provided_password.encode('utf-8'), stored_password.encode('utf-8'))
+
+def register_user(username, password, role, additional_info=None):
+    """Register a new user."""
+    try:
+        # Check if user already exists
+        query = f"SELECT * FROM c WHERE c.id = '{username}'"
+        existing_user = list(container.query_items(query=query, enable_cross_partition_query=True))
+        
+        if existing_user:
+            return False  # User already exists
+        
+        # Create new user document
+        new_user = {
+            'id': username,
+            'password': hash_password(password),
+            'role': role,
+            'additional_info': additional_info or {}
+        }
+        
+        container.create_item(body=new_user)
+        return True
+    except exceptions.CosmosHttpResponseError:
+        return False
+
+def authenticate_user(username, password):
+    """Authenticate a user."""
+    try:
+        query = f"SELECT * FROM c WHERE c.id = '{username}'"
+        results = list(container.query_items(query=query, enable_cross_partition_query=True))
+        
+        if results:
+            stored_user = results[0]
+            if verify_password(stored_user['password'], password):
+                return True
+    except exceptions.CosmosHttpResponseError:
+        pass
+    
+    return False
+
+def get_user_role(username):
+    """Get the role of a user."""
+    try:
+        query = f"SELECT c.role FROM c WHERE c.id = '{username}'"
+        results = list(container.query_items(query=query, enable_cross_partition_query=True))
+        
+        if results:
+            return results[0]['role']
+    except exceptions.CosmosHttpResponseError:
+        pass
+    
+    return None
+
+def update_user_info(username, new_info):
+    """Update user information."""
+    try:
+        query = f"SELECT * FROM c WHERE c.id = '{username}'"
+        results = list(container.query_items(query=query, enable_cross_partition_query=True))
+        
+        if results:
+            user = results[0]
+            user['additional_info'].update(new_info)
+            container.upsert_item(user)
+            return True
+    except exceptions.CosmosHttpResponseError:
+        pass
+    
+    return False
+
+def delete_user(username):
+    """Delete a user."""
+    try:
+        query = f"SELECT * FROM c WHERE c.id = '{username}'"
+        results = list(container.query_items(query=query, enable_cross_partition_query=True))
+        
+        if results:
+            user = results[0]
+            container.delete_item(item=user, partition_key=username)
+            return True
+    except exceptions.CosmosHttpResponseError:
+        pass
+    
+    return False