lucas/s2p_banking_system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move files to folders

Browse Source
This commit is contained in:
Lucas Mathews
2024-06-07 19:13:14 +02:00
parent 2be3ccb830
commit 3979e2ceb8
17 changed files with 5 additions and 1763 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.gitignore vendored
View File

@@ -1,9 +1,9 @@
flask_session/
application/__pycache__/
application/session_data.json
bank.db
test_database.db
log.txt
server/bank.db
server/test_database.db
server/log.txt
# Byte-compiled / optimized / DLL files
__pycache__/

39
api.py
View File

@@ -1,39 +0,0 @@
# Lucas Mathews - Fontys Student ID: 5023572
# Banking System API
###############
### Modules ###
###############
import connexion # Imports connexion module
from config import CONFIG # Imports the configuration file
from manager import * # Imports the Manager file that contains the functions for the API
from flask_session import Session # Imports the session module
#################
### Connexion ###
#################
def create_app():
"""Creates the API using Connexion."""
app = connexion.FlaskApp(__name__)
app.add_api(CONFIG["api_file"]["name"])
flask_app = app.app
flask_app.config['SECRET_KEY'] = CONFIG["sessions"]["secret_key"]
flask_app.config['SESSION_TYPE'] = 'filesystem'
Session(flask_app)
return app
def API():
"""Runs the API."""
app = create_app()
app.run(host=CONFIG["server"]["listen_ip"], port=CONFIG["server"]["port"], debug=CONFIG["server"]["debug"]) # Runs the API using the configuration file
################
### Run Code ###
################
if __name__ == "__main__":
API()

902
api.yml
View File

@@ -1,902 +0,0 @@
openapi: 3.0.3
info:
title: Banking API
description: |-
Lucas Mathews - Fontys Student ID: 5023572
contact:
email: 522499@student.fontys.nl
version: 3.0.0
servers:
- url: http://127.0.0.1:81
tags:
- name: client
description: Operations for Client Accounts
- name: account
description: Operations for Bank Accounts
- name: transaction
description: Operations for Transactions
- name: auth
description: Operations for Authentication
- name: system
description: Operations for System
- name: admin
description: Operations for Admin
paths:
/Client/Login:
post:
tags:
- auth
summary: Log in to the system
description: Log in to the system
operationId: manager.login
requestBody:
description: Login credentials
required: true
content:
application/json:
schema:
type: object
properties:
client_id:
type: string
description: Client ID
client_hash:
type: string
description: Hashed password
required:
- client_id
- client_hash
responses:
'200':
description: Successful operation
content:
application/json:
schema:
type: object
properties:
success:
type: boolean
message:
type: string
'401':
description: Invalid Client ID/password supplied
content:
application/json:
schema:
type: object
properties:
success:
type: boolean
message:
type: string
/Client/Logout:
post:
tags:
- auth
summary: Log out from the system
description: Log out from the system
operationId: manager.logout
responses:
'200':
description: Successful operation
'404':
description: Already logged out
/Client/Status:
get:
tags:
- auth
summary: Get login status
description: Get login status
operationId: manager.status
responses:
'200':
description: Logged in
'400':
description: Not logged in
/Client/Password:
put:
tags:
- auth
summary: Change password
description: Change password
operationId: manager.change_password
requestBody:
required: true
content:
application/json:
schema:
type: object
properties:
client_id:
type: string
description: ID of client to change password
hash_old_password:
type: string
description: Old password as a hash
hash_new_password:
type: string
description: New password as a hash
otp_code:
type: integer
format: int32
description: OTP to verify
required:
- client_id
- hash_old_password
- hash_new_password
- otp_code
responses:
'200':
description: Password changed successfully
'400':
description: Validation error
'401':
description: Unauthorised
'404':
description: Client not found
/OTP/Generate:
post:
tags:
- auth
summary: Generate OTP
description: Generate OTP
operationId: manager.generate_otp
parameters:
- name: client_id
in: query
description: ID of client to generate OTP
required: true
schema:
type: string
responses:
'200':
description: OTP generated
'401':
description: Unauthorised
'400':
description: OTP not valid
'404':
description: client_id not found
/Client:
post:
tags:
- client
summary: Update an existing client
description: Update an existing client Id
operationId: manager.update_client
parameters:
- name: client_id
in: query
description: ID of client to update
required: true
schema:
type: string
- name: otp_code
in: query
description: OTP to verify
required: true
schema:
type: integer
format: int32
- name: name
in: query
description: Client Name
required: false
schema:
type: string
- name: birthdate
in: query
description: Client Birthdate (dd-mm-yyyy)
required: false
schema:
type: string
- name: address
in: query
description: Client Address
required: false
schema:
type: string
- name: phone_number
in: query
description: Client Phone Number
required: false
schema:
type: string
- name: email
in: query
description: Client Email Address
required: false
schema:
type: string
- name: notes
in: query
description: Notes about client
required: false
schema:
type: string
responses:
'200':
description: Successful operation
'404':
description: Invalid Client ID supplied
'403':
description: Unauthorised
'405':
description: OTP not valid
get:
tags:
- client
summary: Get a client by ID
description: Get a client by ID
operationId: manager.get_client
parameters:
- name: client_id
in: query
description: ID of client to return
required: true
schema:
type: string
responses:
'200':
description: Successful operation
content:
application/json:
schema:
$ref: '#/components/schemas/Client'
'400':
description: Invalid Client ID supplied
'404':
description: Client not found
/Client/Accounts:
get:
tags:
- client
summary: Get all accounts for a client
description: Get all accounts for a client
operationId: manager.get_accounts
parameters:
- name: client_id
in: query
description: ID of client to return accounts for
required: true
schema:
type: string
responses:
'200':
description: Successful operation
content:
application/json:
schema:
$ref: '#/components/schemas/ApiResponse'
'400':
description: Invalid input
'404':
description: No accounts found
/Account:
post:
tags:
- account
summary: Add a new account
description: Add a new account to the system
operationId: manager.add_account
parameters:
- name: client_id
in: query
description: ID of client to add account to
required: true
schema:
type: string
- name: description
in: query
description: Account description
required: true
schema:
type: string
- name: account_type
in: query
description: Type of account
required: true
schema:
type: string
- name: notes
in: query
description: Notes about account
required: false
schema:
type: string
responses:
'200':
description: Successful operation
'400':
description: Invalid input
'404':
description: client_id not found
put:
tags:
- account
summary: Update an existing account
description: Update an existing account
operationId: manager.update_account
parameters:
- name: account_id
in: query
description: ID of account to update
required: true
schema:
type: string
- name: otp_code
in: query
description: OTP to verify
required: true
schema:
type: integer
format: int32
- name: description
in: query
description: Account description
required: false
schema:
type: string
- name: account_type
in: query
description: Type of account
required: false
schema:
type: string
- name: notes
in: query
description: Notes about account
required: false
schema:
type: string
responses:
'200':
description: Successful operation
'400':
description: Invalid Account ID supplied
get:
tags:
- account
summary: Get an account by ID
description: Get an account by ID
operationId: manager.get_account
parameters:
- name: account_id
in: query
description: ID of account to return
required: true
schema:
type: string
responses:
'200':
description: Successful operation
content:
application/json:
schema:
$ref: '#/components/schemas/Account'
'400':
description: Invalid Account ID supplied
'404':
description: Account not found
/Transaction:
get:
tags:
- transaction
summary: Get a transaction by ID
description: Get a transaction by ID
operationId: manager.get_transaction
parameters:
- name: transaction_id
in: query
description: ID of transaction to return
required: true
schema:
type: string
responses:
'200':
description: Successful operation
content:
application/json:
schema:
$ref: '#/components/schemas/Transaction'
'400':
description: Invalid Transaction ID supplied
'404':
description: Transaction not found
post:
tags:
- transaction
summary: Add a new transaction
description: Add a new transaction to the system
operationId: manager.add_transaction
parameters:
- name: amount
in: query
description: Amount of transaction
required: true
schema:
type: number
format: float
- name: account_id
in: query
description: Account number the money paid from
required: true
schema:
type: string
- name: recipient_account_id
in: query
description: Recipient account_id
required: true
schema:
type: string
- name: otp_code
in: query
description: OTP to verify
required: true
schema:
type: integer
format: int32
- name: description
in: query
description: Description of transaction
required: false
schema:
type: string
responses:
'200':
description: Successful operation
'400':
description: Invalid input
'403':
description: Invalid OTP
'401':
description: Insufficient funds
'404':
description: account_id not Found
/Transaction/History:
get:
tags:
- transaction
summary: Get transaction history
description: Get transaction history
operationId: manager.transaction_history
parameters:
- name: account_id
in: query
description: ID of account to return
required: true
schema:
type: string
responses:
'200':
description: Successful operation
content:
application/json:
schema:
$ref: '#/components/schemas/Transaction'
'400':
description: Invalid input
'404':
description: No transactions found
/Admin/Interest:
post:
tags:
- admin
summary: Apply interest
description: Apply interest to account
operationId: manager.apply_interest
requestBody:
description: Apply interest to account
content:
application/json:
schema:
$ref: '#/components/schemas/Account'
required: true
responses:
'200':
description: Successful operation
content:
application/json:
schema:
$ref: '#/components/schemas/Account'
'400':
description: Invalid input
'422':
description: Validation exception
put:
tags:
- admin
summary: Apply fee
description: Apply fee to account
operationId: manager.apply_fee
requestBody:
description: Apply fee to account
content:
application/json:
schema:
$ref: '#/components/schemas/Account'
required: true
responses:
'200':
description: Successful operation
content:
application/json:
schema:
$ref: '#/components/schemas/Account'
'400':
description: Invalid input
'422':
description: Validation exception
/Admin/Clients:
get:
tags:
- admin
summary: Get all clients
description: Get all clients
operationId: manager.get_all_clients
responses:
'200':
description: Successful operation
content:
application/json:
schema:
$ref: '#/components/schemas/ApiResponse'
'400':
description: Invalid input
'404':
description: No clients found
/Admin/Accounts:
get:
tags:
- admin
summary: Get all accounts
description: Get all accounts
operationId: manager.get_all_accounts
responses:
'200':
description: Successful operation
content:
application/json:
schema:
$ref: '#/components/schemas/ApiResponse'
'400':
description: Invalid input
'404':
description: No accounts found
/Admin/Transactions:
get:
tags:
- admin
summary: Get all transactions
description: Get all transactions
operationId: manager.get_all_transactions
responses:
'200':
description: Successful operation
content:
application/json:
schema:
$ref: '#/components/schemas/Transaction'
'400':
description: Invalid input
'404':
description: No transactions found
/System/Hash:
get:
tags:
- system
summary: Hash password
description: Pass a string through the hashing algorithm
operationId: manager.hash_password
parameters:
- name: password
in: query
description: Password to hash
required: true
schema:
type: string
responses:
'200':
description: Successful operation
content:
application/json:
schema:
type: string
'400':
description: Invalid input
'401':
description: Unauthorised
/System/Timestamp:
get:
tags:
- system
summary: Get the timestamp
description: Gets the date and time in the appropriate format
operationId: manager.timestamp
responses:
'200':
description: Successful operation
content:
application/json:
schema:
type: string
'400':
description: Invalid input
'401':
description: Unauthorised
/Admin/Balance:
get:
tags:
- admin
summary: Test the balance of all accounts
description: Tests the balance of all accounts and alerts of any discrepancies
operationId: manager.test_account_balances
responses:
'200':
description: Successful operation
content:
application/json:
schema:
type: string
'401':
description: Unauthorised
/System/Initialise:
get:
tags:
- system
summary: Initialise the system
description: Initialises the system with test data
operationId: manager.initialise_database
parameters:
- name: password
in: query
description: Password to initialise the system
required: true
schema:
type: string
- name: email
in: query
description: Email to initialise the system
required: true
schema:
type: string
responses:
'200':
description: Successful operation
'400':
description: Database not empty, this function cannot be used
/Admin/Promote:
put:
tags:
- admin
summary: Promote a client to administrator
description: Promote a client to administrator
operationId: manager.promote_to_admin
parameters:
- name: client_id
in: query
description: ID of client to promote
required: true
schema:
type: string
responses:
'200':
description: Successful operation
'401':
description: Unauthorised
'404':
description: Client not found
/Admin/Demote:
put:
tags:
- admin
summary: Demote a client from administrator
description: Demote a client from administrator
operationId: manager.demote_from_admin
parameters:
- name: client_id
in: query
description: ID of client to demote
required: true
schema:
type: string
responses:
'200':
description: Successful operation
'401':
description: Unauthorised
'404':
description: Client not found
/Admin/Client:
post:
tags:
- admin
summary: Add a new client
description: Add a new client to the system
operationId: manager.add_client
parameters:
- name: name
in: query
description: Client Name
required: true
schema:
type: string
- name: birthdate
in: query
description: Client Birthdate (dd-mm-yyyy)
required: true
schema:
type: string
- name: address
in: query
description: Client Address
required: false
schema:
type: string
- name: phone_number
in: query
description: Client Phone Number
required: true
schema:
type: string
- name: email
in: query
description: Client Email Address
required: true
schema:
type: string
- name: password
in: query
description: Client Password
required: true
schema:
type: string
- name: notes
in: query
description: Notes about client
required: false
schema:
type: string
responses:
'200':
description: "Client created"
'400':
description: Invalid input
'422':
description: Validation exception
/Delete/Client:
delete:
tags:
- admin
summary: Delete a client by ID
description: Delete a client by ID
operationId: manager.delete_client
parameters:
- name: client_id
in: query
description: ID of client to delete
required: true
schema:
type: string
responses:
'200':
description: Successful operation
'400':
description: Invalid Client ID supplied
'404':
description: Client not found
/Delete/Account:
delete:
tags:
- admin
summary: Delete an account by ID
description: Delete an account by ID
operationId: manager.delete_account
parameters:
- name: account_id
in: query
description: ID of account to delete
required: true
schema:
type: string
responses:
'200':
description: Successful operation
'400':
description: Invalid account_id supplied
'404':
description: Account not found
components:
schemas:
ApiResponse:
type: object
properties:
success:
type: boolean
message:
type: string
data:
type: object
example:
success: true
message: "Success"
data: {}
Client:
type: object
properties:
client_id:
type: string
name:
type: string
birthdate:
type: string
opening_timestamp:
type: string
address:
type: string
phone_number:
type: string
email:
type: string
example:
client_id: "896d4ed8"
name: "Lucas Mathews"
birthdate: "21-05-1980"
opening_timestamp: "17-04-2022 16:21:12"
address: "Rachelsmolen 1, 5612MA, Eindhoven"
phone_number: "0612345678"
email: "john.d@fontys.nl"
Account:
type: object
properties:
account_id:
type: string
client_id:
type: string
decription:
type: string
opening_timestamp:
type: string
account_type:
type: string
balance:
type: integer
format: float
notes:
type: string
example:
account_id: "63b6e8e8"
client_id: "896d4ed8"
description: "Savings Account"
opening_timestamp: "17-04-2022 16:21:12"
account_type: Savings Account"
balance: 2314.23
notes: "This account is for savings"
Transaction:
type: object
properties:
transaction_id:
type: string
transaction_type:
type: string
amount:
type: integer
format: float
timestamp:
type: string
description:
type: string
account_to:
type: string
recipient_account_number:
type: string
example:
transaction_id: "d1c4d836-9418-437f-b21c-5cc03d3fdc33"
transaction_type: "Deposit"
amount: 100.00
timestamp: "17-04-2022 16:21:12"
description: "Deposit to Savings Account"
account_id: "896d4ed8"
recipient_account_id: "05225f17"

2
application/app.ini
View File

@@ -1,7 +1,7 @@
[server]
ip = 0.0.0.0
port = 81
url = http://127.0.0.1:81
url = http://0.0.0.0:81
[preferences]
dark_theme = dark

2
application/session_data.json
View File

@@ -1 +1 @@
{"session_cookie": {"session": "I5vE6-7HHC9fAildRSp-bxrPKTaglz4CWQgh1fdoFE4"}, "client_id": "9755c18f"}
{"session_cookie": {"session": "Ax5AB_dcW5bj0zsH6ia9ClsAIJ80MnED8e3dcCc2HAU"}, "client_id": "9755c18f"}

BIN
bank.db
View File

Binary file not shown.

29
bank.ini
View File

@@ -1,29 +0,0 @@
[database]
name=bank.db
[api_file]
name=api.yml
[server]
listen_ip=0.0.0.0
port=81
debug=True
[frontend]
listen_ip=0.0.0.0
port=80
debug=True
[api]
url=http://0.0.0.0:81/
[sessions]
secret_key=57d7dfef5a519fe73d3ba1a9ced6477f
[smtp]
host=email-smtp.us-east-1.amazonaws.com
port=465
username=AKIAYW7TTSMBEUADU2UG
password=BIR8EVWA1bsw+YL8WyAq8BZJZC8vdZrxcrnGe4GnAXEh
sender_name=Luxbank
sender_email=bank@luxdomain.xyz

43
class_account.py
View File

@@ -1,43 +0,0 @@
# Lucas Mathews - Fontys Student ID: 5023572
# Banking System Account Class
from sqlalchemy import ForeignKey, Column, String, Integer, Boolean
from sqlalchemy.orm import relationship
from class_base import Base
class Account(Base):
__tablename__ = 'accounts'
account_id = Column("account_id", String, primary_key=True)
client_id = Column(String, ForeignKey('clients.client_id'))
description = Column("description", String)
open_timestamp = Column("open_timestamp", String)
account_type = Column("account_type", String)
balance = Column("balance", Integer)
enabled = Column("enabled", Boolean)
notes = Column("notes", String)
transactions = relationship("Transaction", foreign_keys='Transaction.account_id', backref="account")
def __init__(self, account_id, client_id, description, open_timestamp, account_type, balance, enabled, notes, transactions):
"""Initialises the account object."""
self.account_id = account_id
self.client_id = client_id
self.description = description
self.open_timestamp = open_timestamp
self.account_type = account_type
self.balance = balance
self.enabled = enabled
self.notes = notes
self.transactions = transactions if transactions is not None else []
def to_dict(self):
"""Returns the account as a dictionary."""
return {
"account_id": self.account_id,
"client_id": self.client_id,
"description": self.description,
"open_timestamp": self.open_timestamp,
"account_type": self.account_type,
"balance": self.balance,
"notes": self.notes
}

6
class_base.py
View File

@@ -1,6 +0,0 @@
# Lucas Mathews - Fontys Student ID: 5023572
# Banking System Base Class
from sqlalchemy.orm import declarative_base
Base = declarative_base()

50
class_client.py
View File

@@ -1,50 +0,0 @@
# Lucas Mathews - Fontys Student ID: 5023572
# Banking System Client Class
from sqlalchemy import Column, String, Boolean
from sqlalchemy.orm import relationship
from class_base import Base
class Client(Base):
__tablename__ = 'clients'
client_id = Column("client_id", String, primary_key=True)
name = Column("name", String)
birthdate = Column("birthdate", String)
opening_timestamp = Column("opening_timestamp", String)
address = Column("address", String)
phone_number = Column("phone_number", String)
email = Column("email", String)
hash = Column("hash", String)
notes = Column("notes", String)
enabled = Column("enabled", Boolean)
administrator = Column("administrator", Boolean)
accounts = relationship("Account", backref="client")
def __init__(self, client_id, name, birthdate, opening_timestamp, address, phone_number, email, hash, notes, enabled, administrator, accounts):
"""Initialises the client object."""
self.client_id = client_id
self.name = name
self.birthdate = birthdate
self.opening_timestamp = opening_timestamp
self.address = address
self.phone_number = phone_number
self.email = email
self.hash = hash
self.notes = notes
self.enabled = enabled
self.administrator = administrator
self.accounts = accounts if accounts is not None else []
def to_dict(self):
"""Returns the client as a dictionary."""
return {
"client_id": self.client_id,
"name": self.name,
"birthdate": self.birthdate,
"opening_timestamp": self.opening_timestamp,
"address": self.address,
"phone_number": self.phone_number,
"email": self.email,
}

38
class_transaction.py
View File

@@ -1,38 +0,0 @@
# Lucas Mathews - Fontys Student ID: 5023572
# Banking System Transaction Class
from sqlalchemy import Column, String, Integer, ForeignKey
from class_base import Base
class Transaction(Base):
__tablename__ = 'transactions'
transaction_id = Column("transaction_id", String, primary_key=True)
transaction_type = Column("transaction_type", String)
amount = Column("amount", Integer)
timestamp = Column("timestamp", String)
description = Column("description", String)
account_id = Column(String, ForeignKey('accounts.account_id'))
recipient_account_id = Column(String, ForeignKey('accounts.account_id'))
def __init__(self, transaction_id, transaction_type, amount, timestamp, description, account_id, recipient_account_id = None):
"""Initialises the Transaction object."""
self.transaction_id = transaction_id
self.transaction_type = transaction_type
self.amount = amount
self.timestamp = timestamp
self.description = description
self.account_id = account_id
self.recipient_account_id = recipient_account_id
def to_dict(self):
"""Converts the Transaction object to a dictionary."""
return {
"transaction_id": self.transaction_id,
"transaction_type": self.transaction_type,
"amount": self.amount,
"timestamp": self.timestamp,
"description": self.description,
"account_id": self.account_id,
"recipient_account_id": self.recipient_account_id
}

7
config.py
View File

@@ -1,7 +0,0 @@
# Lucas Mathews - Fontys Student ID: 5023572
# Banking System Config Parser
import configparser
CONFIG = configparser.ConfigParser()
CONFIG.read("bank.ini")

0
database.py → database/database.py
View File

52
emailer.py
View File

@@ -1,52 +0,0 @@
# Lucas Mathews - Fontys Student ID: 5023572
# Banking System Emailer File
import smtplib
import ssl
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from config import CONFIG
class EmailSendingError(Exception):
"""Custom exception raised when an error occurs during email sending."""
def __init__(self, message, original_error=None):
super().__init__(message)
self.original_error = original_error
def send_email(receiver_email, subject, body):
"""Sends an email to the specified receiver email address."""
sender_email = CONFIG["smtp"]["sender_email"]
message = MIMEMultipart()
message["From"] = f"{CONFIG['smtp']['sender_name']} <{sender_email}>"
message["To"] = receiver_email
message["Subject"] = subject
message.attach(MIMEText(body, "plain"))
text = message.as_string()
context = ssl.create_default_context()
try:
with smtplib.SMTP_SSL(CONFIG["smtp"]["host"], CONFIG["smtp"]["port"], context=context) as server:
server.login(CONFIG["smtp"]["username"], CONFIG["smtp"]["password"])
server.sendmail(sender_email, receiver_email, text)
from manager import log_event
log_event(f"Email '{subject}' sent to {receiver_email}") # Log the message
except Exception as e:
error_message = f"Failed to send email to {receiver_email}: {e}"
from manager import log_event
log_event(error_message) # Log the error
raise EmailSendingError(error_message)
if __name__ == "__main__":
email = "l.mathews@student.fontys.nl"
subject = "Test Email"
body = "This is a test email."
try:
send_email(email, subject, body)
except EmailSendingError:
print("Email sending failed.")

558
manager.py
View File

@@ -1,558 +0,0 @@
# Lucas Mathews - Fontys Student ID: 5023572
# Banking System Manager File
from class_client import Client
from class_account import Account
from class_transaction import Transaction
from emailer import EmailSendingError # Import the EmailSendingError class to handle email sending errors
from flask import jsonify, session as flask_session # Imports the Flask modules
from functools import wraps # For decorators / user login
from database import * # Importing the database connection
from emailer import send_email # Importing the emailer function
from flask import session as flask_session
from flask import request
from database import session
import hashlib # For password hashing
import datetime # For timestamps
import uuid # For unique identifiers
import random # For OTP generation
import time # For OTP generation
import re # For password hash validation
otps = {} # Temporary dictionary to store OTPs and their creation time
##############
### System ###
##############
def timestamp():
"""Returns the current timestamp in the format 'YYYY-MM-DD HH:MM:SS'."""
return (datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
def hash_password(password:str):
"""Hashes a password using the SHA-512 algorithm and returns the hexadecimal representation of the hash."""
return hashlib.sha512(password.encode()).hexdigest()
def generate_uuid():
"""Generates a unique identifier using the UUID4 algorithm and returns it as a string."""
return str(uuid.uuid4())
def generate_uuid_short():
"""Generates a short unique identifier using the UUID4 algorithm and returns the first 8 characters as a string."""
return str(uuid.uuid4())[:8]
def get_email(client_id:str):
"""Returns the email of a client given their client_id. If the client is not found, returns None."""
client = session.query(Client).filter_by(client_id=client_id).one_or_none()
return client.email if client else None
def format_response(success: bool, message: str = '', data: dict = None):
"""Formats the response for the API so that it is standardised across all functions."""
response = {
'success': success,
'message': message,
}
if data is not None:
response['data'] = data
return jsonify(response)
def get_current_client():
"""Returns the client_id and administrator status of the currently logged in client. If no client is logged in, returns None, None."""
if 'client_id' not in flask_session:
return None, None
client = flask_session['client_id']
client_obj = session.query(Client).filter_by(client_id=client).one_or_none()
if client_obj is None:
return None, None
return client_obj.client_id, client_obj.administrator
def verify_otp(client_id:str, otp:int):
"""Verifies a one time password for a client. Returns True if the OTP is correct and False otherwise."""
if client_id in otps:
stored_otp, creation_time = otps[client_id]
if stored_otp == otp and time.time() - creation_time <= 300: # Check if OTP is within 5 minutes
return True
return False
def delete_otp(client_id:str):
"""Deletes the OTP for a client."""
if client_id in otps:
del otps[client_id]
def clean_expired_otps():
"""Checks for expired OTPs and deletes them. An OTP is considered expired if it is older than 5 minutes."""
current_time = time.time()
expired_otps = [client_id for client_id, (otp, creation_time) in otps.items() if current_time - creation_time > 300] # Find OTPs older than 5 minutes
otps_removed = 0
for client_id in expired_otps:
delete_otp(client_id)
otps_removed += 1
log_event(f"Cleaned {otps_removed} expired OTPs.")
def log_event(data_to_log:str):
"""Logs an event to the log file."""
with open("log.txt", "a") as log_file:
log_file.write(f"{timestamp()} - {data_to_log}\n")
######################
### Authentication ###
######################
def login():
"""Logs in a client using their client_id and password hash. Returns a success message if the login is successful and an error message otherwise."""
data = request.get_json()
client_id = data.get('client_id')
client_hash = data.get('client_hash')
client = session.query(Client).filter_by(client_id=client_id).first()
if client and client.hash == client_hash:
flask_session['client_id'] = client_id
log_event(f"{client_id} logged in successfully.")
return format_response(True, f"{flask_session['client_id']} logged in successfully."), 200
return format_response(False, "Invalid client_id or password."), 401
def logout():
"""Logs out a client. Returns a success message if the logout is successful and an error message otherwise."""
if 'client_id' in flask_session:
flask_session.pop('client_id', None)
return format_response(True, "Logged out successfully."), 200
return format_response(False, "Not logged in."), 400
def status():
"""Returns the current status of the client."""
if 'client_id' in flask_session:
return format_response(True, f"Logged in as {flask_session['client_id']}"), 200
else:
return format_response(False, "Not logged in."), 400
def login_required(f):
"""Decorator function to check if a client is logged in before accessing a route."""
@wraps(f)
def decorated_function(*args, **kwargs):
if 'client_id' not in flask_session:
return format_response(False, "Not logged in."), 401
return f(*args, **kwargs)
return decorated_function
def admin_required(f):
"""Decorator function to check if a client is an administrator before accessing a route."""
@wraps(f)
def decorated_function(*args, **kwargs):
if 'client_id' not in flask_session:
return format_response(False, "Not logged in."), 401
for client in session.query(Client).all():
if client.client_id == flask_session['client_id']:
if client.administrator == 1:
return f(*args, **kwargs)
return format_response(False, "Not authorised."), 403
return decorated_function
@login_required
def generate_otp(client_id: str):
"""Generates a one-time password for a client and sends it to their email address. Returns a success message if the OTP is generated and an error message otherwise."""
current_client_id, is_admin = get_current_client()
if not is_admin and client_id != current_client_id:
return format_response(False, "You can only generate OTPs for your own client account."), 403
email = get_email(client_id)
if email:
password = int(random.randint(100000, 999999)) # Generate a 6-digit OTP
try:
send_email(email, "Luxbank One Time Password", f"Your one-time password is: {password}")
otps[client_id] = (password, time.time()) # Store the OTP and the current time
log_event(f"OTP Code {password} emailed to {email}")
print(f"OTP Code {password} emailed to {email}")
return format_response(True, "OTP generated and sent successfully."), 200
except EmailSendingError as e:
log_event(f"Error sending email: {str(e)}")
error_message = "Error sending email. Please try again later."
if e.original_error:
error_message += f" Original error: {str(e.original_error)}"
return format_response(False, error_message), 500
else:
return format_response(False, "Email address not found for the client."), 404
##############
### Client ###
##############
@login_required
def get_client(client_id: str):
"""Returns a specific client in the database. If the client is not found, returns an error message."""
current_client_id, is_admin = get_current_client()
if not is_admin and client_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403
client = session.query(Client).filter_by(client_id=client_id).one_or_none()
if client:
return format_response(True, "", client.to_dict()), 200
return format_response(False, "Client not found."), 404
@login_required
def update_client(client_id:str, otp_code:int, **kwargs):
"""Updates a client in the database. If the client is not found, returns an error message."""
current_client_id, is_admin = get_current_client()
if not verify_otp(current_client_id, otp_code):
return format_response(False, "Invalid OTP."), 400 # Changed to 400 Bad Request
if not is_admin and client_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403
client = session.query(Client).filter_by(client_id=client_id).first()
if client:
for field in ['name', 'birthdate', 'address', 'phone_number', 'email', 'notes']:
if field in kwargs and kwargs[field] is not None:
setattr(client, field, kwargs[field])
session.commit()
return format_response(True, f"Client ID: {client_id} has been updated."), 200
return format_response(False, "Client not found."), 404
def change_password():
"""Changes the password for a client in the database. If the client is not found, returns an error message."""
data = request.get_json()
client_id = data.get('client_id')
hash_old_password = data.get('hash_old_password')
hash_new_password = data.get('hash_new_password')
otp_code = data.get('otp_code')
current_client_id, is_admin = get_current_client()
otp_verified = verify_otp(client_id, otp_code) # Verify if the OTP is correct
if not is_admin and client_id != current_client_id: # Check if the client is authorized to change the password
return format_response(False, "You can only update your own client information."), 401
if not otp_verified: # Recheck OTP verification after authorisation check
return format_response(False, "Invalid OTP."), 400
hash_format = r'^[0-9a-f]{128}$' # Validate new password format
if not re.match(hash_format, hash_new_password):
return format_response(False, "Invalid new password format (must be provided as a hash)."), 400
client = session.query(Client).filter_by(client_id=client_id).first() # Check if the old password hash matches and update to the new password hash
if client:
if client.hash == hash_old_password:
client.hash = hash_new_password
session.commit()
delete_otp(current_client_id)
log_event(f"Password for client_id {client_id} has been updated by {current_client_id}.")
return format_response(True, f"Password for client_id {client_id} has been updated."), 200
else:
return format_response(False, "Invalid old password."), 400
else:
return format_response(False, "Client not found."), 404
@login_required
def get_accounts(client_id: str):
"""Returns all accounts for a specific client in the database. If the client is not found, returns an error message."""
current_client_id, is_admin = get_current_client()
if current_client_id is None:
raise Exception("No current client found")
if not is_admin and client_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403
accounts = session.query(Account).filter(Account.client_id == client_id)
return format_response(True, "", [account.to_dict() for account in accounts]), 200
###############
### Account ###
###############
@login_required
def get_account(account_id: str):
"""Returns a specific account in the database. If the account is not found, returns an error message."""
current_client_id, is_admin = get_current_client()
account = session.query(Account).filter_by(account_id=account_id).one_or_none()
if account is None:
return format_response(False, "Account not found."), 404
account_owner = account.client_id
if not is_admin and account_owner != current_client_id:
return format_response(False, "You can only view your own client information."), 403
return format_response(True, "", account.to_dict()), 200
@login_required
def add_account(client_id:str, description:str, account_type:str, **kwargs):
"""Adds a new account to the database. If the client is not found, returns an error message."""
current_client_id, is_admin = get_current_client()
if not is_admin and client_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403
account_id = generate_uuid_short()
notes = kwargs.get("notes", None)
client_found = None
for client in session.query(Client).all(): # Find the client
if client.client_id == client_id:
client_found = client
break
if client_found is None: # If the client is not found, return an error
return format_response(False, "Client not found."), 404
new_account = Account(account_id, client_id, description, timestamp(), account_type, 0, 1, notes, None) # Create the new account
session.add(new_account)
session.commit()
return format_response(True, f"New account has been added: account_id: {account_id}"), 200
@login_required
def update_account(account_id: str, otp_code: str, **kwargs):
"""Updates an account in the database. If the account is not found, returns an error message."""
current_client_id, is_admin = get_current_client()
# Verify OTP
if not verify_otp(current_client_id, otp_code):
return format_response(False, "Invalid OTP."), 400
# Query the account once
account = session.query(Account).filter_by(account_id=account_id).one_or_none()
if account is None:
return format_response(False, "Account not found."), 404
account_owner = account.client_id
# Check permissions
if not is_admin and account_owner != current_client_id:
return format_response(False, "You can only update your own account information."), 403
# Update the account with provided kwargs
description = kwargs.get("description")
account_type = kwargs.get("account_type")
balance = kwargs.get("balance")
enabled = kwargs.get("enabled")
notes = kwargs.get("notes")
if description is not None:
account.description = description
if account_type is not None:
account.account_type = account_type
if balance is not None:
account.balance = balance
if enabled is not None:
account.enabled = enabled
if notes is not None:
account.notes = notes
# Commit the changes
session.commit()
return format_response(True, f"account_id: {account_id} has been updated."), 200
###################
### Transaction ###
###################
@login_required
def get_transaction(transaction_id:int):
"""Returns a specific transaction in the database. If the transaction is not found, returns an error message."""
current_client_id, is_admin = get_current_client()
transaction = session.query(Transaction).filter_by(transaction_id=transaction_id).one_or_none()
if not transaction:
return format_response(False, "Transaction not found."), 404
account = session.query(Account).filter_by(account_id=transaction.account_id).one_or_none()
recipient_account = session.query(Account).filter_by(account_id=transaction.recipient_account_id).one_or_none()
if not is_admin and (account.client_id != current_client_id and recipient_account.client_id != current_client_id):
return format_response(False, "You can only view your own client information."), 403
return format_response(True, "", transaction.to_dict()), 200
@login_required
def add_transaction(amount: float, account_id: str, recipient_account_id: str, otp_code: int, description: str):
"""Adds a new transaction to the database. If the account is not found, returns an error message."""
print(f"Adding transaction: amount: {amount}, account_id: {account_id}, recipient_account_id: {recipient_account_id}, otp_code: {otp_code}, description: {description}")
current_client_id, is_admin = get_current_client()
if not is_admin and account_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403
otp_verified = verify_otp(current_client_id, otp_code)
if not otp_verified:
return format_response(False, "Invalid OTP."), 400
transaction_id = generate_uuid()
account_from = session.query(Account).filter_by(account_id=account_id).one_or_none()
account_dest = session.query(Account).filter_by(account_id=recipient_account_id).one_or_none()
if account_from is None or account_dest is None:
return format_response(False, "Account not found."), 404
if account_from.balance < amount:
return format_response(False, "Insufficient funds."), 400
delete_otp(current_client_id)
# Perform the transaction
account_from.balance -= amount
account_dest.balance += amount
transaction_type = "transfer"
session.commit()
# Create the transaction record
new_transaction = Transaction(transaction_id, transaction_type, amount, timestamp(), description, account_id, recipient_account_id)
session.add(new_transaction)
session.commit()
return format_response(True, f"New transaction has been added: transaction_id: {transaction_id}"), 200
@login_required
def transaction_history(account_id:int):
"""Returns all transactions for a specific account in the database. If the account is not found, returns an error message."""
current_client_id, is_admin = get_current_client()
account = session.query(Account).filter_by(account_id=account_id).one_or_none()
if not account:
return format_response(False, "Account not found."), 404
if not is_admin and account.client_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403
result = session.query(Transaction).filter(Transaction.account_id == account_id)
return format_response(True, "", [transaction.to_dict() for transaction in result]), 200
#####################
### Administrator ###
#####################
@admin_required
def delete_client(client_id:str):
"""Deletes a client from the database. If the client is not found, returns an error message."""
if client_id == flask_session['client_id']:
return format_response(False, "You cannot delete your own account."), 400
for client in session.query(Client).all():
if client.client_id == client_id:
if client.accounts == None:
session.delete(client)
session.commit()
log_event(f"Client ID: {client_id} has been removed by {flask_session['client_id']}.")
return format_response(True, f"client_id: {client_id} has been removed."), 200
else:
return format_response(False, "Client has accounts and can not be removed."), 400
return format_response(False, "Client not found."), 404
@admin_required
def delete_account(account_id:str):
"""Deletes an account from the database. If the account is not found, returns an error message."""
for account in session.query(Account).all():
if account.account_id == account_id:
if account.balance == 0:
session.delete(account)
session.commit()
log_event(f"Account ID: {account_id} has been removed by {flask_session['client_id']}.")
return format_response(True, f"account_id: {account_id} has been removed."), 200
else:
return format_response(False, "Account has a balance and can not be removed."), 400
return format_response(False, "Account not found."), 404
@admin_required
def get_all_clients():
"""Returns all clients in the database."""
clients = session.query(Client).all()
log_event(f"All clients have been retrieved by {flask_session['client_id']}.")
return format_response(True, "", [client.to_dict() for client in clients]), 200
@admin_required
def get_all_accounts():
"""Returns all accounts in the database."""
accounts = session.query(Account).all()
log_event(f"All accounts have been retrieved by {flask_session['client_id']}.")
return format_response(True, "", [account.to_dict() for account in accounts]), 200
@admin_required
def get_all_transactions():
"""Returns all transactions in the database."""
transactions = session.query(Transaction).all()
log_event(f"All transactions have been retrieved by {flask_session['client_id']}.")
return format_response(True, "", [transaction.to_dict() for transaction in transactions]), 200
@admin_required
def apply_interest(account_id:int, interest_rate:float):
"""Applies interest to an account based on the interest rate. If the account is not found, returns an error message."""
for account in session.query(Account).filter(Account.account_id == account_id):
if account.account_id == account_id:
interest = account.balance * interest_rate
account.balance += interest
session.commit()
log_event(f"Interest of €{interest} has been applied to Account ID: {account_id} by {flask_session['client_id']}.")
return format_response(True, f"€{interest} in interest has been applied to Account ID: {account_id}."), 200
return format_response(False, "Account not found."), 404
@admin_required
def apply_fee(account_id:int, fee:float):
"""Applies a fee to an account based on the fee amount. If the account is not found, returns an error message."""
for account in session.query(Account).all():
if account.account_id == account_id:
account.balance -= fee
session.commit()
log_event(f"Fee of €{fee} has been applied to Account ID: {account_id} by {flask_session['client_id']}.")
return format_response(True, f"€{fee} in fees has been applied to Account ID: {account_id}."), 200
return format_response(False, "Account not found."), 404
@admin_required
def delete_transaction(transaction_id:int):
"""Deletes a transaction from the database. If the transaction is not found, returns an error message."""
for transaction in session.query(Transaction).all():
if transaction.transaction_id == transaction_id:
session.delete(transaction)
session.commit()
log_event(f"Transaction ID: {transaction_id} has been removed by {flask_session['client_id']}.")
return format_response(True, f"Transaction ID: {transaction_id} has been removed."), 200
return format_response(False, "Transaction not found."), 404
@admin_required
def modify_balance(transaction_id:int, amount:int):
"""Modifies the amount of a transaction in the database. If the transaction is not found, returns an error message."""
for transaction in session.query(Transaction).all():
if transaction.transaction_id == transaction_id:
transaction.amount = amount
session.commit()
log_event(f"Transaction ID: {transaction_id} has been modified by {flask_session['client_id']}.")
return format_response(True, f"Transaction ID: {transaction_id} has been modified."), 200
return format_response(False, "Transaction not found."), 404
@admin_required
def test_account_balances():
"""Checks all account balances in the database and returns a list of discrepancies."""
log_event(f"Account balances have been checked by {flask_session['client_id']}.")
all_transactions = session.query(Transaction).all()# Get all transactions from the database
calculated_balances = {} # Initialize a dictionary to store the calculated balance for each account
for transaction in all_transactions: # Go through each transaction
if transaction.account_id not in calculated_balances: # If the account ID of the transaction is not in the dictionary, add it with a balance of 0
calculated_balances[transaction.account_id] = 0
if transaction.transaction_type == 'Deposit': # Update the calculated balance for the account
calculated_balances[transaction.account_id] += transaction.amount
elif transaction.transaction_type == 'Withdrawal':
calculated_balances[transaction.account_id] -= transaction.amount
all_accounts = session.query(Account).all() # Get all accounts from the database
discrepancies = [] # Initialize a list to store the discrepancies
for account in all_accounts: # Go through each account
if calculated_balances.get(account.account_id, 0) != account.balance: # If the calculated balance doesn't match the stored balance, add the discrepancy to the list
discrepancies.append({"error": f"Alert: Account {account.account_id} has a balance discrepancy. Stored balance is {account.balance}, but calculated balance is {calculated_balances.get(account.account_id, 0)}."})
return format_response(True, "", discrepancies), 200 # Return the list of discrepancies
@admin_required
def add_client(name:str, birthdate:str, address:str, phone_number:str, email:str, password:str, **kwargs):
"""Adds a new client to the database."""
client_id = generate_uuid_short()
notes = kwargs.get("notes", None)
new_client = Client(client_id, name, birthdate, timestamp(), address, phone_number, email, hash_password(password), notes, 1, 0, None)
session.add(new_client)
session.commit()
log_event(f"New client has been added: client_id: {client_id} by {flask_session['client_id']}.")
return format_response(True, f"New client has been added: client_id: {client_id}"), 200
def initialise_database(password:str, email:str):
"""Initialises the database with an administrator client if no clients exist."""
existing_clients = session.query(Client).all() # Check if any clients exist in the database
if not existing_clients: # If no clients exist, create an administrator client
new_client = Client(generate_uuid_short(), "ADMIN", "ADMIN", timestamp(), "ADMIN", "ADMIN", email, hash_password(password), None, 1, 0, None)
session.add(new_client)
session.commit()
admin_client = session.query(Client).filter_by(name='ADMIN').one() # Retrieve the administrator client
admin_client.administrator = 1 # Set the new client as an administrator
session.commit()
log_event(f"Database initialised with administrator account with client_id {admin_client.client_id}.")
return format_response(True, f"Database initialised with administrator account with client_id {admin_client.client_id}"), 200
return format_response(False, "Database not empty."), 400
@admin_required
def promote_to_admin(client_id:str):
"""Promotes a client to administrator status. If the client is not found, returns an error message."""
for client in session.query(Client).all():
if client.client_id == client_id:
client.administrator = 1
session.commit()
log_event(f"Client ID: {client_id} has been promoted to administrator by {flask_session['client_id']}.")
return format_response(True, f"client_id: {client_id} has been promoted to administrator."), 200
return format_response(False, f"client_id: {client_id} is not found."), 404
@admin_required
def demote_from_admin(client_id:str):
"""Demotes a client from administrator status. If the client is not found, returns an error message."""
for client in session.query(Client).all():
if client.client_id == client_id:
client.administrator = 0
session.commit()
log_event(f"Client ID: {client_id} has been demoted from administrator by {flask_session['client_id']}.")
return format_response(True, f"client_id: {client_id} has been demoted from administrator."), 200
return format_response(False, f"client_id: {client_id} is not found."), 404

8
requirements.txt
View File

@@ -1,8 +0,0 @@
flask
connexion[swagger-ui]==2.14.2
requests
sqlalchemy
flask-session
faker
customtkinter
schedule

26
scheduler.py
View File

@@ -1,26 +0,0 @@
# Lucas Mathews - Fontys Student ID: 5023572
# Banking System Scheduler
import schedule
from manager import log_event
def run_schedule():
while True:
schedule.run_pending()
#################
### Functions ###
#################
def clean_otp():
"""Cleans the OTP table."""
from manager import clean_expired_otps
removed_otps = clean_expired_otps()
log_event(f"Removed {removed_otps} expired OTPs.")
#################
### Schedules ###
#################
schedule.every(2).seconds.do(clean_otp)