Feat: \n 1. Contact Us page form submission success page created. \n 2. Contact Us page styling and CAPTCHA text content. \n 3. Removal of ERP, Google CAPTCHA, and ALTCHA API code and left over comments in JavaScript, Python.
This commit is contained in:
@@ -12,10 +12,12 @@ Initializes the Flask application, sets the configuration based on the environme
|
||||
|
||||
# IMPORTS
|
||||
# internal
|
||||
from business_objects.api import API
|
||||
from datastores.datastore_base import DataStore_Base
|
||||
from forms.contact import Form_Contact
|
||||
from helpers.helper_app import Helper_App
|
||||
from models.model_view_contact import Model_View_Contact
|
||||
from models.model_view_contact_success import Model_View_Contact_Success
|
||||
from models.model_view_home import Model_View_Home
|
||||
import lib.argument_validation as av
|
||||
# external
|
||||
@@ -51,49 +53,26 @@ def contact():
|
||||
form = Form_Contact()
|
||||
model = Model_View_Contact(form)
|
||||
html_body = render_template('pages/core/_contact.html', model = model)
|
||||
return html_body
|
||||
except Exception as e:
|
||||
return jsonify(error=str(e)), 403
|
||||
return html_body
|
||||
return API.get_standard_response(
|
||||
success = False,
|
||||
status_code = 500,
|
||||
message = f"Error: {e}",
|
||||
data = None,
|
||||
errors = [str(e)],
|
||||
meta = None
|
||||
)
|
||||
|
||||
@routes_core.route(Model_View_Contact.HASH_PAGE_CONTACT, methods=['POST'])
|
||||
def contact_post():
|
||||
try:
|
||||
form = Form_Contact()
|
||||
Helper_App.console_log(f"Form submitted: {request.form}")
|
||||
Helper_App.console_log(f"ALTCHA data in request: {request.form.get('altcha')}")
|
||||
if form.validate_on_submit():
|
||||
try:
|
||||
email = form.email.data
|
||||
# CC = form.CC.data # not in use
|
||||
contact_name = form.contact_name.data
|
||||
company_name = form.company_name.data
|
||||
message = form.message.data
|
||||
receive_marketing = form.receive_marketing.data
|
||||
receive_marketing_text = "I would like to receive marketing emails." if receive_marketing else ""
|
||||
# send email
|
||||
mailItem = Message("PARTS Website Contact Us Message", recipients=[current_app.config['MAIL_CONTACT_PUBLIC']])
|
||||
mailItem.body = f"Dear Lord Edward Middleton-Smith,\n\n{message}\n{receive_marketing_text}\nKind regards,\n{contact_name}\n{company_name}\n{email}"
|
||||
mail.send(mailItem)
|
||||
flash('Thank you for your message. We will get back to you soon!', 'success')
|
||||
return "Submitted."
|
||||
except Exception as e:
|
||||
return f"Error: {e}"
|
||||
print(f"Form validation errors: {form.errors}")
|
||||
return "Invalid. Failed to submit."
|
||||
# html_body = render_template('pages/core/_contact.html', model = model)
|
||||
except Exception as e:
|
||||
return jsonify(error=str(e)), 403
|
||||
|
||||
@routes_core.route(Model_View_Contact.HASH_ALTCHA_CREATE_CHALLENGE, methods=['GET'])
|
||||
@routes_core.route(Model_View_Contact.HASH_GET_ALTCHA_CHALLENGE, methods=['GET'])
|
||||
def create_altcha_challenge():
|
||||
Helper_App.console_log(f'secret key: {current_app.app_config.ALTCHA_SECRET_KEY}')
|
||||
options = ChallengeOptions(
|
||||
expires = datetime.datetime.now() + datetime.timedelta(hours=1),
|
||||
max_number = 100000, # The maximum random number
|
||||
hmac_key = current_app.app_config.ALTCHA_SECRET_KEY,
|
||||
)
|
||||
challenge = create_challenge(options)
|
||||
print("Challenge created:", challenge)
|
||||
Helper_App.console_log(f"Challenge created: {challenge}")
|
||||
# return jsonify({"challenge": challenge})
|
||||
return jsonify({
|
||||
"algorithm": challenge.algorithm,
|
||||
@@ -102,58 +81,64 @@ def create_altcha_challenge():
|
||||
"signature": challenge.signature,
|
||||
})
|
||||
|
||||
"""
|
||||
def verify_altcha_signature(payload):
|
||||
"" "Verify the ALTCHA signature"" "
|
||||
if 'algorithm' not in payload or 'signature' not in payload or 'verificationData' not in payload:
|
||||
return False
|
||||
|
||||
algorithm = payload['algorithm']
|
||||
signature = payload['signature']
|
||||
verification_data = payload['verificationData']
|
||||
|
||||
# Calculate SHA hash of the verification data
|
||||
if algorithm == 'SHA-256':
|
||||
hash_func = hashlib.sha256
|
||||
else:
|
||||
# Fallback to SHA-256 if algorithm not specified
|
||||
hash_func = hashlib.sha256
|
||||
|
||||
# Calculate the hash of verification_data
|
||||
data_hash = hash_func(verification_data.encode('utf-8')).digest()
|
||||
|
||||
# Calculate the HMAC signature
|
||||
calculated_signature = hmac.new(
|
||||
current_app.config["ALTCHA_SECRET_KEY"].encode('utf-8'),
|
||||
data_hash,
|
||||
hash_func
|
||||
).hexdigest()
|
||||
|
||||
# Compare the calculated signature with the provided signature
|
||||
return hmac.compare_digest(calculated_signature, signature)
|
||||
@routes_core.route(Model_View_Contact.HASH_PAGE_CONTACT, methods=['POST'])
|
||||
def contact_post():
|
||||
try:
|
||||
form = Form_Contact()
|
||||
if form.validate_on_submit():
|
||||
try:
|
||||
email = form.email.data
|
||||
# CC = form.CC.data # not in use
|
||||
contact_name = form.contact_name.data
|
||||
company_name = form.company_name.data
|
||||
message = form.message.data
|
||||
receive_marketing = form.receive_marketing.data
|
||||
receive_marketing_text = "I would like to receive marketing emails.\n" if receive_marketing else ""
|
||||
# send email
|
||||
mailItem = Message("PARTS Website Contact Us Message", recipients=[current_app.config['MAIL_CONTACT_PUBLIC']])
|
||||
mailItem.body = f"Dear Lord Edward Middleton-Smith,\n\n{message}\n{receive_marketing_text}\nKind regards,\n{contact_name}\n{company_name}\n{email}"
|
||||
mail.send(mailItem)
|
||||
return redirect(url_for(Model_View_Contact.ENDPOINT_PAGE_CONTACT_SUCCESS))
|
||||
except Exception as e:
|
||||
return API.get_standard_response(
|
||||
success = False,
|
||||
status_code = 500,
|
||||
message = f"Error: {e}",
|
||||
data = None,
|
||||
errors = [str(e)],
|
||||
meta = None
|
||||
)
|
||||
return API.get_standard_response(
|
||||
success = False,
|
||||
status_code = 500,
|
||||
message = f"Error: {form.errors}",
|
||||
data = None,
|
||||
errors = [str(form.errors)],
|
||||
meta = None
|
||||
)
|
||||
# html_body = render_template('pages/core/_contact.html', model = model)
|
||||
except Exception as e:
|
||||
return API.get_standard_response(
|
||||
success = False,
|
||||
status_code = 500,
|
||||
message = f"Error: {e}",
|
||||
data = None,
|
||||
errors = [str(e)],
|
||||
meta = None
|
||||
)
|
||||
|
||||
|
||||
|
||||
def create_altcha_dummy_signature(challenge):
|
||||
# Example payload to verify
|
||||
payload = {
|
||||
"algorithm": challenge.algorithm,
|
||||
"challenge": challenge.challenge,
|
||||
"number": 12345, # Example number
|
||||
"salt": challenge.salt,
|
||||
"signature": challenge.signature,
|
||||
}
|
||||
return payload
|
||||
|
||||
@routes_core.route(Model_View_Contact.HASH_ALTCHA_VERIFY_SOLUTION, methods=['POST'])
|
||||
def verify_altcha_challenge():
|
||||
payload = request.json
|
||||
|
||||
ok, err = verify_solution(payload, current_app.config["ALTCHA_SECRET_KEY"], check_expires=True)
|
||||
if err:
|
||||
return jsonify({"error": err}), 400
|
||||
elif ok:
|
||||
return jsonify({"verified": True})
|
||||
else:
|
||||
return jsonify({"verified": False}), 403
|
||||
"""
|
||||
@routes_core.route(Model_View_Contact.HASH_PAGE_CONTACT_SUCCESS, methods=['GET'])
|
||||
def contact_success():
|
||||
try:
|
||||
model = Model_View_Contact_Success()
|
||||
html_body = render_template('pages/core/_contact_success.html', model = model)
|
||||
return html_body
|
||||
except Exception as e:
|
||||
return API.get_standard_response(
|
||||
success = False,
|
||||
status_code = 500,
|
||||
message = f"Error: {e}",
|
||||
data = None,
|
||||
errors = [str(e)],
|
||||
meta = None
|
||||
)
|
||||
|
||||
@@ -1,31 +0,0 @@
|
||||
"""
|
||||
Project: PARTS Website
|
||||
Author: Edward Middleton-Smith
|
||||
Precision And Research Technology Systems Limited
|
||||
|
||||
Technology: App Routing
|
||||
Feature: Store Routes
|
||||
|
||||
Description:
|
||||
Initializes the Flask application, sets the configuration based on the environment, and defines two routes (/ and /about) that render templates with the specified titles.
|
||||
"""
|
||||
|
||||
# IMPORTS
|
||||
# internal
|
||||
from models.model_view_store import Model_View_Store
|
||||
# external
|
||||
from flask import Flask, render_template, jsonify, request, render_template_string, send_from_directory, redirect, url_for, session, Blueprint, current_app, Response
|
||||
from extensions import db, oauth
|
||||
from urllib.parse import quote_plus, urlencode
|
||||
from authlib.integrations.flask_client import OAuth
|
||||
from authlib.integrations.base_client import OAuthError
|
||||
from urllib.parse import quote, urlparse, parse_qs
|
||||
|
||||
routes_store = Blueprint('routes_store', __name__)
|
||||
|
||||
@routes_store.route(Model_View_Store.HASH_SCRIPTS_SECTION_STORE, methods=['GET'])
|
||||
def scripts_section_store():
|
||||
hash_page_current = request.args.get('hash_page_current', default = Model_View_Store.HASH_SCRIPTS_SECTION_STORE, type = str)
|
||||
model = Model_View_Store(hash_page_current=hash_page_current)
|
||||
template = render_template('js/sections/store.js', model = model)
|
||||
return Response(template, mimetype='application/javascript')
|
||||
@@ -1,243 +0,0 @@
|
||||
"""
|
||||
Project: PARTS Website
|
||||
Author: Edward Middleton-Smith
|
||||
Precision And Research Technology Systems Limited
|
||||
|
||||
Technology: App Routing
|
||||
Feature: User Routes
|
||||
|
||||
Description:
|
||||
Initializes the Flask application, sets the configuration based on the environment, and defines two routes (/ and /about) that render templates with the specified titles.
|
||||
"""
|
||||
|
||||
# IMPORTS
|
||||
# internal
|
||||
from models.model_view_base import Model_View_Base
|
||||
from models.model_view_user import Model_View_User
|
||||
from business_objects.user import User, Parameters_User
|
||||
from datastores.datastore_user import DataStore_User
|
||||
from helpers.helper_app import Helper_App
|
||||
import lib.argument_validation as av
|
||||
# external
|
||||
from flask import Flask, render_template, jsonify, request, render_template_string, send_from_directory, redirect, url_for, session, Blueprint, current_app
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
from sqlalchemy import exc
|
||||
from flask_wtf.csrf import generate_csrf
|
||||
from werkzeug.exceptions import BadRequest
|
||||
from extensions import oauth # db,
|
||||
from urllib.parse import quote_plus, urlencode
|
||||
from authlib.integrations.flask_client import OAuth
|
||||
from authlib.integrations.base_client import OAuthError
|
||||
from urllib.parse import quote, urlparse, parse_qs
|
||||
from functools import wraps
|
||||
|
||||
db = SQLAlchemy()
|
||||
|
||||
routes_user = Blueprint('routes_user', __name__)
|
||||
|
||||
def handle_db_disconnect(f):
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
try:
|
||||
return f(*args, **kwargs)
|
||||
except exc.OperationalError as e:
|
||||
if "MySQL server has gone away" in str(e):
|
||||
# Close the session and create a new connection
|
||||
db.session.remove()
|
||||
db.session.rollback()
|
||||
# Retry the operation
|
||||
return f(*args, **kwargs)
|
||||
raise
|
||||
return decorated_function
|
||||
|
||||
# User authentication
|
||||
@routes_user.route("/login", methods=['POST', 'OPTIONS'])
|
||||
def login():
|
||||
try:
|
||||
Helper_App.console_log('login')
|
||||
Helper_App.console_log(f'method={request.method}')
|
||||
try:
|
||||
data = request.json
|
||||
try:
|
||||
data = request.get_json()
|
||||
except:
|
||||
data = {}
|
||||
except:
|
||||
data = {}
|
||||
Helper_App.console_log(f'data={data}')
|
||||
hash_callback = data.get(Model_View_Base.FLAG_CALLBACK, Model_View_Base.HASH_PAGE_HOME)
|
||||
Helper_App.console_log(f'hash_callback: {hash_callback}')
|
||||
|
||||
"""
|
||||
# Verify CSRF token manually
|
||||
Helper_App.console_log(f'request headers={request.headers}')
|
||||
token = request.headers.get(Model_View_Base.FLAG_CSRF_TOKEN)
|
||||
Helper_App.console_log(f'token={token}')
|
||||
Helper_App.console_log(f'session={session}')
|
||||
Helper_App.console_log(f'session token={session.get('csrf_token')}')
|
||||
if not token or token != session.get('csrf_token'):
|
||||
token = data.get(Model_View_Base.FLAG_CSRF_TOKEN, None)
|
||||
Helper_App.console_log(f'token={token}')
|
||||
if not token or token != session.get('csrf_token'):
|
||||
raise BadRequest('Invalid or missing CSRF token')
|
||||
"""
|
||||
# OAuth login
|
||||
# callback_login = F'{Model_View_Base.HASH_CALLBACK_LOGIN}{data.get(Model_View_Base.FLAG_CALLBACK, Model_View_Base.HASH_PAGE_HOME)}'
|
||||
|
||||
# encoded_path = quote(data.get(Model_View_Base.FLAG_CALLBACK, Model_View_Base.HASH_PAGE_HOME))
|
||||
uri_redirect = url_for('routes_user.login_callback', _external=True) # , subpath=encoded_path
|
||||
|
||||
# uri_redirect = f'{current_app.URL_HOST}/login_callback?subpath={data.get(Model_View_Base.FLAG_CALLBACK, Model_View_Base.HASH_PAGE_HOME)}'
|
||||
Helper_App.console_log(f'redirect uri: {uri_redirect}')
|
||||
|
||||
Helper_App.console_log(f'Before red')
|
||||
|
||||
red = oauth.auth0.authorize_redirect(
|
||||
redirect_uri = uri_redirect,
|
||||
state = quote(hash_callback)
|
||||
)
|
||||
Helper_App.console_log(f'redirect: {red}')
|
||||
headers = red.headers['Location']
|
||||
Helper_App.console_log(f'headers: {headers}')
|
||||
parsed_url = urlparse(headers)
|
||||
query_params = parse_qs(parsed_url.query)
|
||||
Helper_App.console_log(f"""
|
||||
OAuth Authorize Redirect URL:
|
||||
|
||||
Base URL: {parsed_url.scheme}://{parsed_url.netloc}{parsed_url.path}
|
||||
{parsed_url}
|
||||
|
||||
Query Parameters: {query_params}
|
||||
""")
|
||||
return jsonify({'Success': True, Model_View_Base.FLAG_STATUS: Model_View_Base.FLAG_SUCCESS, f'{Model_View_Base.FLAG_CALLBACK}': headers})
|
||||
|
||||
return jsonify({'status': 'success', 'redirect': callback})
|
||||
except Exception as e:
|
||||
return jsonify({'status': 'error', 'message': str(e)}), 400
|
||||
|
||||
|
||||
@routes_user.route("/login_callback") # <path:subpath>/<code>
|
||||
@handle_db_disconnect
|
||||
def login_callback():
|
||||
Helper_App.console_log('login_callback')
|
||||
try:
|
||||
error_state = request.args.get(Model_View_User.FLAG_ERROR_OAUTH)
|
||||
has_error = error_state is not None
|
||||
if has_error:
|
||||
error_description = request.args.get(Model_View_User.FLAG_ERROR_DESCRIPTION_OAUTH)
|
||||
error_text = f'Error: {error_state}: {error_description}'
|
||||
Helper_App.console_log(error_text)
|
||||
return login()
|
||||
# Helper_App.console_log(f'code: {code}')
|
||||
token = None
|
||||
try:
|
||||
token = oauth.auth0.authorize_access_token()
|
||||
except Exception as e:
|
||||
# Log the error for debugging
|
||||
Helper_App.console_log(f"Error: {str(e)}")
|
||||
session[current_app.config['ID_TOKEN_USER']] = token
|
||||
# import user id
|
||||
"""
|
||||
Helper_App.console_log(f'str(type(token)) = {str(type(token))}')
|
||||
Helper_App.console_log(f'token = {token}')
|
||||
userinfo = token.get('userinfo')
|
||||
Helper_App.console_log(f'user info: {userinfo}')
|
||||
# id_user = token.get('sub')
|
||||
id_user = userinfo.get('sub')
|
||||
Helper_App.console_log(f'user ID: {id_user}')
|
||||
"""
|
||||
user = User.from_json_auth0(token) # datastore_user.get_user_auth0()
|
||||
Helper_App.console_log(f'user: {user}')
|
||||
filters = Parameters_User.from_user(user)
|
||||
datastore_user = DataStore_User()
|
||||
users, errors = datastore_user.get_many_user(filters, user)
|
||||
try:
|
||||
user = users[0]
|
||||
Helper_App.console_log('User logged in')
|
||||
Helper_App.console_log(f'user ({str(type(user))}): {user}')
|
||||
Helper_App.console_log(f'user key: {Model_View_Base.FLAG_USER}')
|
||||
user_json = user.to_json()
|
||||
session[Model_View_Base.FLAG_USER] = user_json
|
||||
Helper_App.console_log(f'user stored on session')
|
||||
except:
|
||||
Helper_App.console_log(f'User not found: {Parameters_User}\nDatabase query error: {errors}')
|
||||
|
||||
try:
|
||||
hash_callback = token.get('hash_callback')
|
||||
if hash_callback is None:
|
||||
Helper_App.console_log('hash is none')
|
||||
state = request.args.get('state')
|
||||
Helper_App.console_log(f'state: {state}')
|
||||
hash_callback = state # .get('hash_callback')
|
||||
Helper_App.console_log(f'hash_callback: {hash_callback}')
|
||||
except:
|
||||
Helper_App.console_log("get hash callback failed")
|
||||
# id_user = get_id_user()
|
||||
# add user to database
|
||||
# DataStore_Store().add_new_user(id_user) # this is part of get basket - should occur on page load
|
||||
|
||||
Helper_App.console_log(f'user session: {session[Model_View_Base.FLAG_USER]}')
|
||||
return redirect(f"{current_app.config['URL_HOST']}{hash_callback}")
|
||||
except Exception as e:
|
||||
return jsonify({Model_View_Base.FLAG_STATUS: Model_View_Base.FLAG_FAILURE, Model_View_Base.FLAG_MESSAGE: f'Controller error.\n{e}'})
|
||||
|
||||
@routes_user.route("/logout")
|
||||
def logout():
|
||||
session.clear()
|
||||
url_logout = f"https://{current_app.config['DOMAIN_AUTH0']}/v2/logout?" + urlencode(
|
||||
{
|
||||
"returnTo": url_for("routes_user.logout_callback", _external=True),
|
||||
"client_id": current_app.config['ID_AUTH0_CLIENT'],
|
||||
}# ,
|
||||
# quote_via=quote_plus,
|
||||
)
|
||||
Helper_App.console_log(f"Redirecting to {url_logout}")
|
||||
return redirect(url_logout)
|
||||
|
||||
@routes_user.route("/logout_callback") # <path:subpath>/<code>
|
||||
@handle_db_disconnect
|
||||
def logout_callback():
|
||||
return redirect(url_for('routes_core.home'))
|
||||
try:
|
||||
session[current_app.ID_TOKEN_USER] = None
|
||||
user = User()
|
||||
try:
|
||||
hash_callback = token.get('hash_callback')
|
||||
if hash_callback is None:
|
||||
Helper_App.console_log('hash is none')
|
||||
state = request.args.get('state')
|
||||
Helper_App.console_log(f'state: {state}')
|
||||
hash_callback = state # .get('hash_callback')
|
||||
Helper_App.console_log(f'hash_callback: {hash_callback}')
|
||||
except:
|
||||
Helper_App.console_log("get hash callback failed")
|
||||
# id_user = get_id_user()
|
||||
# add user to database
|
||||
# DataStore_Store().add_new_user(id_user) # this is part of get basket - should occur on page load
|
||||
|
||||
Helper_App.console_log(f'user session: {session[Model_View_Base.FLAG_USER]}')
|
||||
return redirect(f'{current_app.URL_HOST}{hash_callback}')
|
||||
except Exception as e:
|
||||
return jsonify({Model_View_Base.FLAG_STATUS: Model_View_Base.FLAG_FAILURE, Model_View_Base.FLAG_MESSAGE: f'Controller error.\n{e}'})
|
||||
|
||||
|
||||
@routes_user.route("/user")
|
||||
def user():
|
||||
try:
|
||||
model = Model_View_User()
|
||||
for currency in model.currencies:
|
||||
if currency.id_currency == model.user.id_currency_default:
|
||||
model.user.currency_default = currency
|
||||
break
|
||||
for region in model.regions:
|
||||
if region.id_region == model.user.id_region_default:
|
||||
model.user.region_default = region
|
||||
break
|
||||
model.users = [model.user]
|
||||
if not model.is_user_logged_in:
|
||||
# return redirect(url_for('routes_user.login', data = jsonify({ Model_View_User.FLAG_CALLBACK: Model_View_User.HASH_PAGE_USER_ACCOUNT })))
|
||||
return redirect(url_for('routes_core.home'))
|
||||
html_body = render_template('pages/user/_user.html', model = model)
|
||||
except Exception as e:
|
||||
return str(e)
|
||||
return html_body
|
||||
Reference in New Issue
Block a user