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

Finish implementing new transacitions

Browse Source
This commit is contained in:
Lucas Mathews
2024-06-07 17:31:56 +02:00
parent 1e95d2b605
commit 21fe74853c
7 changed files with 135 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
api.yml
View File

@@ -416,7 +416,7 @@ paths:
description: Amount of transaction
required: true
schema:
type: integer
type: number
format: float
- name: account_id
in: query
@@ -430,12 +430,6 @@ paths:
required: true
schema:
type: string
- name: description
in: query
description: Description of transaction
required: false
schema:
type: string
- name: otp_code
in: query
description: OTP to verify
@@ -443,6 +437,12 @@ paths:
schema:
type: integer
format: int32
- name: description
in: query
description: Description of transaction
required: false
schema:
type: string
responses:
'200':
description: Successful operation

BIN
application/__pycache__/connection.cpython-312.pyc
View File

Binary file not shown.

28
application/connection.py
View File

@@ -169,20 +169,38 @@ def update_account(account_id, otp_code:int, description=None, notes=None):
print(f"RequestException: {e}")
return {'success': False, 'message': "Could not connect to the server. Please try again later."}
def new_transaction(ammount, account_id, recipient_account_id, otp_code, description):
def new_transaction(account_id, description, amount, recipient_account_id, otp_code):
"""Creates a new transaction for the given account."""
try:
with open('application\\session_data.json', 'r') as f:
session_data = json.load(f)
if description == "":
description = None
params = {"account_id": account_id, "recipient_account_id": recipient_account_id, "amount": ammount, "description": description, "otp_code": otp_code}
# Prepare the query parameters
params = {
"amount": amount,
"account_id": account_id,
"recipient_account_id": recipient_account_id,
"otp_code": otp_code,
"description": description}
print(f"Params: {params}")
response = requests.post(CONFIG["server"]["url"] + "/Transaction", cookies=session_data['session_cookie'], params=params)
response.raise_for_status()
print(f"Response: {response.json()}")
return response.json()
except requests.exceptions.HTTPError as e:
if e.response.status_code == 404:
return {'success': False, 'message': 'Account not found.'}
elif e.response.status_code == 403:
return {'success': False, 'message': 'Invalid OTP code.'}
elif e.response.status_code == 400:
return {'success': False, 'message': 'Invalid request. Please check the OTP code and other details.'}
else:
print(f"HTTPError: {e}")
return {'success': False, 'message': f"Unexpected error occurred. Error: {e}"}
except requests.exceptions.RequestException as e:
print(f"RequestException: {e}")
return {'success': False, 'message': "Could not connect to the server. Please try again later."}
return {'success': False, 'message': f"Could not connect to the server. Error: {e}"}
def generate_otp():
"""Generates a new OTP for the current client, which is sent by Email."""

86
application/new_transaction.py
View File

@@ -24,64 +24,64 @@ else:
def submit_transaction():
"""Submit a new transaction to the server."""
account_id = account_combobox.get()
recipient_id = recipient_text.get("1.0", "end-1c")
amount = amount_text.get("1.0", "end-1c")
description = description_text.get("1.0", "end-1c")
if not amount or not recipient_id or not account_id:
messagebox.showerror("Error", "Please fill in all fields.")
amount = amount_text.get("1.0", "end").strip()
account_id = account_combobox.get().strip()
recipient_id = recipient_text.get("1.0", "end").strip()
description = description_text.get("1.0", "end").strip()
otp_code:int = otp_text.get("1.0", "end").strip()
if not amount or not recipient_id or not account_id or not otp_code:
messagebox.showerror("Error", "Please fill in manditory fields.")
return
# Convert amount to float
try:
if verify_accounts(account_id, recipient_id, amount) is False:
return
if not otp_code.isdigit() or len(otp_code) != 6:
messagebox.showerror("Error", "Please enter a valid 6-digit OTP code.")
return
try: # Convert amount to float
amount = float(amount)
except ValueError:
messagebox.showerror("Error", "Invalid amount entered.")
return
account_verification = verify_accounts()
if account_verification is False:
messagebox.showerror("Error", "Could not verify account IDs.")
try: # Convert otp code to int
otp_code = int(otp_code)
except ValueError:
messagebox.showerror("Error", "Invalid OTP code entered.")
return
response = new_transaction(account_id, description, amount, recipient_id)
if response is None or "data" not in response:
response = new_transaction(account_id, description, amount, recipient_id, otp_code)
if response is None or "success" not in response:
messagebox.showerror("Error", "Could not submit transaction.")
return
transaction_id = response["data"]
otp = generate_otp(account_id, transaction_id)
if otp is None or "data" not in otp:
messagebox.showerror("Error", "Could not generate OTP.")
if not response['success']: # Check if the transaction was unsuccessful
messagebox.showerror("Error", response['message']) # Show the error message in a popup
return
messagebox.showinfo("Success", f"Transaction submitted successfully. OTP: {otp['data']}")
if response['success']: # Check if the transaction was successful
messagebox.showinfo("Success", "Transaction submitted successfully.")
root.destroy() # Close the window
return
def verify_accounts():
"""Verify that the account IDs are valid."""
try:
account_id = account_combobox.get()
recipient_id = recipient_text.get("1.0", "end-1c")
account = get_account(account_id)
recipient_id = get_account(recipient_id)
except requests.exceptions.RequestException as e:
messagebox.showerror("Error", f"Failed to get account details: {e}")
def verify_accounts(account_id, recipient_id, amount):
"""Verify that both account IDs are valid, and the from account has enough balance to make the transaction."""
account = get_account(account_id)
if account is None or "data" not in account or "balance" not in account["data"]:
messagebox.showerror("Error", "Invalid account ID.")
return False
if account is None or recipient_id is None or "data" not in account or "data" not in recipient_id:
messagebox.showerror("Error", "Server did not return the expected response.")
recipient = get_account(recipient_id)
if recipient is None or "data" not in recipient:
messagebox.showerror("Error", "Invalid recipient ID.")
return False
if "account_id" not in account["data"]:
messagebox.showerror("Error", "Account ID not found.")
try: # Convert amount to float
amount = float(amount)
except ValueError:
messagebox.showerror("Error", "Invalid amount entered.")
return False
if "account_id" not in recipient_id["data"]:
messagebox.showerror("Error", "Recipient Account ID not found.")
return False
amount = amount_text.get("1.0", "end-1c")
if account["data"]["balance"] < amount: # Check the client has the required balance
messagebox.showerror("Error", "Insufficient funds.")
return False
messagebox.showinfo("Success", "Accounts verified successfully.")
account_balance = account["data"]["balance"]
if account_balance < amount:
messagebox.showerror("Error", "Insufficient balance.")
return False
return True
def populate_accounts(client_id):
@@ -156,7 +156,7 @@ otp_text = customtkinter.CTkTextbox(root, height=2, width=250)
otp_text.pack(pady=5)
# Create the submit button
submit_button = customtkinter.CTkButton(root, text="Submit", command=submit_transaction)
submit_button = customtkinter.CTkButton(root, text="Verify & Submit", command=submit_transaction)
submit_button.pack(pady=5)
# Populate accounts combobox with the given client_id

2
application/session_data.json
View File

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

BIN
bank.db
View File

Binary file not shown.

103
manager.py
View File

@@ -228,7 +228,7 @@ def change_password():
if client.hash == hash_old_password:
client.hash = hash_new_password
session.commit()
delete_otp(client_id)
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:
@@ -252,17 +252,17 @@ def get_accounts(client_id: str):
###############
@login_required
def get_account(account_id:str):
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_owner = session.query(Account).filter_by(account_id=account_id).one_or_none().client_id
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
account = session.query(Account).filter_by(account_id=account_id).one_or_none()
for account in session.query(Account).all():
if account.account_id == account_id:
return format_response(True, "", account.to_dict()), 200
return format_response(False, "Account not found."), 404
return format_response(True, "", account.to_dict()), 200
@login_required
def add_account(client_id:str, description:str, account_type:str, **kwargs):
@@ -285,34 +285,47 @@ def add_account(client_id:str, description:str, account_type:str, **kwargs):
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):
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
account_owner = session.query(Account).filter_by(account_id=account_id).one_or_none().client_id
# 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 view your own client information."), 403
for account in session.query(Account).all():
if account.account_id == account_id:
description = kwargs.get("description", None)
account_type = kwargs.get("account_type", None)
balance = kwargs.get("balance", None)
enabled = kwargs.get("enabled", None)
notes = kwargs.get("notes", None)
if description:
account.description = description
if account_type:
account.account_type = account_type
if balance:
account.balance = balance
if enabled:
account.enabled = enabled
if notes:
account.notes = notes
session.commit()
return format_response(True, f"account_id: {account_id} has been updated."), 200
return format_response(False, "Account not found."), 404
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 ###
@@ -332,31 +345,37 @@ def get_transaction(transaction_id:int):
return format_response(True, "", transaction.to_dict()), 200
@login_required
def add_transaction(amount:int, account_id, recipient_account_id, otp_code:int, **kwargs):
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) # Verify if the OTP is correct
otp_verified = verify_otp(current_client_id, otp_code)
if not otp_verified:
return format_response(False, "Invalid OTP."), 400
transaction_id = generate_uuid()
for account in session.query(Account).all():
if account.account_id == account_id:
account_from = account
if account.account_id == recipient_account_id:
account_dest = account
if account_from.balance < amount: # Check if account has enough funds
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
account_from.balance -= amount # Perform the transaction
delete_otp(current_client_id)
# Perform the transaction
account_from.balance -= amount
account_dest.balance += amount
transaction_type = "transfer"
session.commit()
description = kwargs.get("description", None) # Create the transaction record
# 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
return format_response(True, f"New transaction has been added: transaction_id: {transaction_id}"), 200
@login_required
def transaction_history(account_id:int):