jeanGaston/RF-AD
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add docstrings

Browse Source
This commit is contained in:
Paul Gonçalves Monnet 2024-06-07 11:25:02 +02:00
parent c1a624010c
commit 9b8588f7eb
4 changed files with 278 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
Server/Program/Webserver.py
View File

@ -118,7 +118,6 @@ def door_access():
access_granted, upn = check_access(rfid_uid, door_id)
if access_granted:
print("")
log_access_attempt(DBFILE, upn, rfid_uid, True, door_id)
return jsonify({"access_granted": True, "upn": upn}), 200
@ -128,10 +127,24 @@ def door_access():
def run_flask_app():
"""
Run the Flask web application.
This function starts the Flask web application with debugging enabled,
no reloader, on the specified port and host. It serves as the main entry
point for running the web server.
"""
app.run(debug=True, use_reloader=False, port=WebServerPORT, host="0.0.0.0")
def run_webServer_thread():
"""
Start the Flask web server in a separate thread.
This function initializes and starts a new thread to run the Flask web
application. It allows the web server to run concurrently with other
tasks in the main program, ensuring the web interface remains responsive.
"""
print(f"STARTING WEB SERVER ON PORT {WebServerPORT}")
flask_thread = Thread(target=run_flask_app, daemon=True)
flask_thread.start()

180
Server/Program/database.py
View File

@ -5,6 +5,18 @@ from env import DBFILE
# Function to check if a table exists in the database
def table_exists(cursor, table_name):
"""
Check if a table exists in the database.
This function checks whether a table with the specified name exists in the database.
## Parameters:
- cursor (sqlite3.Cursor): The cursor object to execute SQL queries.
- table_name (str): The name of the table to check.
## Returns:
- bool: True if the table exists, False otherwise.
"""
cursor.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,)
)
@ -13,6 +25,14 @@ def table_exists(cursor, table_name):
# Function to create the Users table
def create_users_table(cursor):
"""
Create the Users table in the database.
This function creates the Users table with columns for user principal name (upn), RFID UID, and member of groups.
## Parameters:
- cursor (sqlite3.Cursor): The cursor object to execute SQL queries.
"""
cursor.execute("""CREATE TABLE Users (
upn TEXT PRIMARY KEY,
rFIDUID TEXT,
@ -23,6 +43,14 @@ def create_users_table(cursor):
# Function to create the Groups table
def create_groups_table(cursor):
"""
Create the Groups table in the database.
This function creates the Groups table with a single column for common name (cn) of the group.
## Parameters:
- cursor (sqlite3.Cursor): The cursor object to execute SQL queries.
"""
cursor.execute("""CREATE TABLE Groups (
cn TEXT PRIMARY KEY
)""")
@ -30,6 +58,14 @@ def create_groups_table(cursor):
# Function to create the Doors table
def create_doors_table(cursor):
"""
Create the Doors table in the database.
This function creates the Doors table with columns for door ID and associated group common name.
## Parameters:
- cursor (sqlite3.Cursor): The cursor object to execute SQL queries.
"""
cursor.execute("""CREATE TABLE Doors (
id INTEGER PRIMARY KEY,
GroupCn TEXT,
@ -40,9 +76,13 @@ def create_doors_table(cursor):
# Function to create the logs table
def create_logs_table(cursor):
"""
Create a log table with columns id, timestamp, user, and granted.
Create the logs table in the database.
:param db_file: The database file path.
This function creates the logs table with columns for ID (auto-incremented), timestamp, user, RFID UID, door ID,
and access granted status. Foreign key constraints are set on the door ID, user, and RFID UID columns.
## Parameters:
- cursor (sqlite3.Cursor): The cursor object to execute SQL queries.
"""
cursor.execute("""
CREATE TABLE IF NOT EXISTS log (
@ -61,6 +101,19 @@ def create_logs_table(cursor):
# Function to setup the database
def setup_database(db_file):
"""
Set up the SQLite database by creating necessary tables if they don't already exist.
This function checks if the Users, Groups, Doors, and Log tables exist in the database. If any of them don't exist,
it creates them using their respective creation functions. After creating or verifying the tables, it commits
the changes and closes the database connection.
## Parameters:
- db_file (str): The file path to the SQLite database.
## Returns:
- None
"""
# Connect to the SQLite database
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
@ -98,13 +151,20 @@ def setup_database(db_file):
def log_access_attempt(db_file, user, rFIDUID, granted, doorID):
"""
Log an access attempt to the log table.
Log an access attempt to the database.
:param db_file: The database file path.
:param user: The user attempting access.
:param rFIDUID: The user's tag uid
:param granted: Whether access was granted (True/False).
:param doorID: The door id
This function inserts a new entry into the log table of the SQLite database, recording details about the access attempt,
such as the timestamp, user, RFID UID, whether access was granted, and the door ID.
## Parameters:
- db_file (str): The file path to the SQLite database.
- user (str): The user's UPN (User Principal Name).
- rFIDUID (str): The RFID UID associated with the access attempt.
- granted (bool): A boolean indicating whether access was granted (True) or denied (False).
- doorID (int): The ID of the door where the access attempt occurred.
# Returns:
- None
"""
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
@ -122,6 +182,12 @@ def log_access_attempt(db_file, user, rFIDUID, granted, doorID):
def print_users_table(cursor):
"""
Print the content of the Users table.
## Parameters:
- cursor (sqlite3.Cursor): Cursor object for executing SQLite queries.
"""
cursor.execute("SELECT * FROM Users")
rows = cursor.fetchall()
print("Users:")
@ -131,6 +197,12 @@ def print_users_table(cursor):
# Function to print the content of the Groups table
def print_groups_table(cursor):
"""
Print the content of the Groups table.
## Parameters:
- cursor (sqlite3.Cursor): Cursor object for executing SQLite queries.
"""
cursor.execute("SELECT * FROM Groups")
rows = cursor.fetchall()
print("Groups:")
@ -140,6 +212,12 @@ def print_groups_table(cursor):
# Function to print the content of the Doors table
def print_doors_table(cursor):
"""
Print the content of the Doors table.
## Parameters:
- cursor (sqlite3.Cursor): Cursor object for executing SQLite queries.
"""
cursor.execute("SELECT * FROM Doors")
rows = cursor.fetchall()
print("Doors:")
@ -149,6 +227,12 @@ def print_doors_table(cursor):
# Function to print the content of the Log table
def print_log_table(cursor):
"""
Print the content of the Log table.
## Parameters:
- cursor (sqlite3.Cursor): Cursor object for executing SQLite queries.
"""
cursor.execute("SELECT * FROM log")
rows = cursor.fetchall()
print("Logs:")
@ -158,6 +242,12 @@ def print_log_table(cursor):
# Function to print the content of the entire database
def print_database_content(db_file):
"""
Print the content of the entire database.
## Parameters:
- db_file (str): The file path to the SQLite database.
"""
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
@ -172,7 +262,9 @@ def print_database_content(db_file):
def get_logs():
"""
Fetch all logs from the log table in the database.
:return: List of log records.
## Returns:
- list: List of log records.
"""
conn = sqlite3.connect(DBFILE)
cursor = conn.cursor()
@ -193,8 +285,12 @@ def get_latest_logs(db_file, limit=10):
"""
Fetch the latest logs from the database.
:param limit: The number of latest logs to fetch.
:return: List of log entries.
## Parameters:
- db_file (str): The file path to the SQLite database.
- limit (int): The number of latest logs to fetch. Default is 10.
## Returns:
- list: List of log entries.
"""
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
@ -216,6 +312,15 @@ def get_latest_logs(db_file, limit=10):
# Function to fetch list of existing groups from the database
def get_existing_groups(db_file):
"""
Fetches a list of existing groups from the database.
## Parameters:
- db_file (str): The file path to the SQLite database.
## Returns:
- list: List of existing group names.
"""
try:
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
@ -229,6 +334,15 @@ def get_existing_groups(db_file):
def delete_group_from_database(group_cn):
"""
Delete a group from the database.
This function deletes a group with the specified common name (cn) from both the Groups and Doors tables
in the database.
## Parameters:
- group_cn (str): The common name of the group to delete.
"""
conn = sqlite3.connect(DBFILE)
cursor = conn.cursor()
cursor.execute("DELETE FROM Groups WHERE cn = ?", (group_cn,))
@ -238,6 +352,14 @@ def delete_group_from_database(group_cn):
def get_doors():
"""
Retrieve all doors from the database.
This function fetches all rows from the Doors table in the database and returns them as a list of tuples.
## Returns:
- list: A list of tuples representing door records.
"""
conn = sqlite3.connect(DBFILE)
cursor = conn.cursor()
cursor.execute("SELECT * FROM Doors")
@ -249,7 +371,9 @@ def get_doors():
def get_users():
"""
Fetch all users from the Users table in the database.
:return: List of user records.
## Returns:
- list: List of user records.
"""
conn = sqlite3.connect(DBFILE)
cursor = conn.cursor()
@ -263,6 +387,22 @@ def get_users():
# Function to add a door to the database
def add_door_to_database(db_file, group_cn, Door_id):
"""
Add a door to the database.
This function inserts a new door record into the Doors table with the specified group common name (cn)
and door ID.
## Parameters:
- db_file (str): The file path to the SQLite database.
- group_cn (str): The common name of the group associated with the door.
- Door_id (int): The ID of the door.
## Returns:
- bool: True if the door was added successfully, False otherwise.
## Raises:
- sqlite3.Error: If there's an error executing the SQL query.
"""
try:
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
@ -285,6 +425,22 @@ def add_door_to_database(db_file, group_cn, Door_id):
# Function to verify if the user is allowed to open the door
def check_access(rfid_uid_str, door_id):
"""
Check if the user is allowed to open the door.
This function verifies if the user associated with the given RFID UID is allowed to open the door
specified by the door ID.
## Parameters:
- rfid_uid_str (str): The RFID UID of the user.
- door_id (int): The ID of the door.
## Returns:
- tuple: A tuple containing a boolean value indicating access permission and the user's UPN
if access is granted, otherwise (False, None).
## Raises:
- sqlite3.Error: If there's an error executing the SQL query.
"""
try:
conn = sqlite3.connect(DBFILE) # Update with your database file path
cursor = conn.cursor()

113
Server/Program/ldapSync.py
View File

@ -9,12 +9,14 @@ from env import DOOR_ACCESS_GROUPS_DN, LDAPPASS, LDAPUSER, LDAP_SERVER, USERS_DN
# Function to initialize LDAP connection
def initialize_ldap_connection():
"""
## Settings :
None
## Behavior
Init the connection to the LDAP server.
Return LDAPobjet instance when connected
if it fail, return None and print error code
Initialize the LDAP connection.
This function attempts to establish a connection to the LDAP server using the provided server address,
user credentials, and settings. If the connection is successful, it returns the connection object.
In case of an error, it prints the error and returns None.
## Returns:
- ldap.LDAPObject or None: The LDAP connection object if successful, otherwise None.
"""
try:
connect = ldap.initialize(LDAP_SERVER)
@ -30,12 +32,18 @@ def initialize_ldap_connection():
# Function to retrieve users from LDAP
def retrieve_users_from_ldap(ldap_connection):
"""
## Settings :
- ldap_connection : LDAPobjet instance
## Behavior
retrieve the users in the specified OU
Return result when it success
if it fail, return empty list and print error code
Retrieve users from LDAP.
This function searches the LDAP directory for users within the specified base DN and returns the search result.
It searches for objects with the 'user' object class within the subtree of the specified base DN.
## Parameters:
- ldap_connection (ldap.LDAPObject): The LDAP connection object.
## Returns:
- list of tuple: A list of tuples containing the search result, where each tuple represents a user entry.
Each tuple consists of the DN (Distinguished Name) of the user entry and its attributes.
Returns an empty list if an error occurs during the LDAP search.
"""
try:
result = ldap_connection.search_s(
@ -50,12 +58,18 @@ def retrieve_users_from_ldap(ldap_connection):
# Function to retrieve groups from LDAP
def retrieve_groups_from_ldap(ldap_connection):
"""
## Settings :
- ldap_connection : LDAPobjet instance
## Behavior
retrieve the groups in the specified OU
Return result when it success
if it fail, return empty list and print error code
Retrieve groups from LDAP.
This function searches the LDAP directory for groups within the specified base DN and returns the search result.
It searches for objects with the 'group' object class within the subtree of the specified base DN.
## Parameters:
- ldap_connection (ldap.LDAPObject): The LDAP connection object.
## Returns:
- list of tuple: A list of tuples containing the search result, where each tuple represents a group entry.
Each tuple consists of the DN (Distinguished Name) of the group entry and its attributes.
Returns an empty list if an error occurs during the LDAP search.
"""
try:
result = ldap_connection.search_s(
@ -69,6 +83,26 @@ def retrieve_groups_from_ldap(ldap_connection):
# Function to add user to the database or update if already exists
def add_user_to_database(conn, cursor, upn, rfid_uid, member_of):
"""
Add a user to the database or update the user's information if they already exist.
This function checks if a user with the given UPN (User Principal Name) already exists in the database.
If the user exists and their RFID UID or group membership has changed, the function updates the user's
record. If the user does not exist, the function inserts a new record for the user.
## Parameters:
- conn (sqlite3.Connection): The SQLite database connection.
- cursor (sqlite3.Cursor): The cursor object for executing SQL queries.
- upn (str): The User Principal Name of the user.
- rfid_uid (str): The RFID UID associated with the user.
- member_of (str): The group membership (CN) of the user.
## Returns:
- None
## Raises:
- sqlite3.Error: If an error occurs while accessing the SQLite database.
"""
try:
cursor.execute("SELECT * FROM Users WHERE upn=?", (upn,))
existing_user = cursor.fetchone()
@ -99,6 +133,24 @@ def add_user_to_database(conn, cursor, upn, rfid_uid, member_of):
# Function to add group to the database or update if already exists
def add_group_to_database(conn, cursor, cn):
"""
Add a group to the database if it does not already exist.
This function checks if a group with the given CN (Common Name) already exists in the database.
If the group exists, it prints a message indicating that the group already exists. If the group
does not exist, it inserts a new record for the group.
Parameters:
conn (sqlite3.Connection): The SQLite database connection.
cursor (sqlite3.Cursor): The cursor object for executing SQL queries.
cn (str): The Common Name of the group.
Returns:
None
Raises:
sqlite3.Error: If an error occurs while accessing the SQLite database.
"""
try:
cursor.execute("SELECT * FROM Groups WHERE cn=?", (cn,))
existing_group = cursor.fetchone()
@ -184,10 +236,35 @@ def sync_ldap_to_database(db_file):
def run_sync_ldap_to_database_thread(db_file):
"""
Run the LDAP synchronization process in a separate thread.
This function initiates the synchronization of LDAP data to the database in a background thread.
It ensures that the LDAP synchronization runs asynchronously, allowing the main program to continue
running without being blocked.
Parameters:
db_file (str): The path to the SQLite database file.
Returns:
None
"""
print(f"[{datetime.now()}] Running LDAP sync")
threading.Thread(target=sync_ldap_to_database, args=(db_file,), daemon=True).start()
def schedule_sync_ldap_to_database(db_file):
"""
Schedule the LDAP synchronization process to run immediately and then every 5 minutes.
This function runs the LDAP synchronization process in a background thread immediately upon invocation
and sets up a recurring schedule to run the synchronization every 5 minutes.
Parameters:
db_file (str): The path to the SQLite database file.
Returns:
None
"""
run_sync_ldap_to_database_thread(db_file) # Run immediately
schedule.every(5).minutes.do(run_sync_ldap_to_database_thread, db_file)

3
Server/Program/server.py
View File

@ -1,12 +1,11 @@
from ldapSync import schedule_sync_ldap_to_database
from database import setup_database, print_database_content
from database import setup_database
from Webserver import run_webServer_thread
from env import DBFILE
import schedule
setup_database(DBFILE)
# print_database_content(DBFILE)
run_webServer_thread()
schedule_sync_ldap_to_database(DBFILE)