modification of code and loggger apply and changes of update from.
This commit is contained in:
@@ -7,7 +7,12 @@ class FileHandler:
|
||||
|
||||
@staticmethod
|
||||
def CHeckExistingOrCreateNewUploadFolder():
|
||||
#Wheteher path exists
|
||||
# Whether path exists
|
||||
os.makedirs(FileHandler.UPLOAD_FOLDER, exist_ok=True)
|
||||
return
|
||||
|
||||
|
||||
@staticmethod
|
||||
def CheckExistingOrCreateNewLoggerFolder():
|
||||
if not os.path.exists("logs"):
|
||||
os.mkdir("logs")
|
||||
return
|
||||
@@ -1,22 +1,49 @@
|
||||
import os
|
||||
from flask import Flask, render_template, request, redirect, url_for, send_from_directory, flash, jsonify, json
|
||||
from flask import current_app
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
|
||||
from flask import session, request, current_app
|
||||
|
||||
|
||||
class LogHelper:
|
||||
|
||||
@staticmethod
|
||||
def setup_logger(app):
|
||||
if not os.path.exists("logs"):
|
||||
os.makedirs("logs")
|
||||
|
||||
formatter = logging.Formatter("%(asctime)s | %(levelname)s | User:%(user)s | IP:%(ip)s | %(message)s")
|
||||
|
||||
file_handler = logging.FileHandler("logs/app.log")
|
||||
file_handler.setLevel(logging.INFO)
|
||||
file_handler.setFormatter(formatter)
|
||||
|
||||
stream_handler = logging.StreamHandler()
|
||||
stream_handler.setLevel(logging.INFO)
|
||||
stream_handler.setFormatter(formatter)
|
||||
|
||||
app.logger.setLevel(logging.INFO)
|
||||
app.logger.addHandler(file_handler)
|
||||
app.logger.addHandler(stream_handler)
|
||||
|
||||
# ---------------------------------------
|
||||
# Log User Activity
|
||||
# ---------------------------------------
|
||||
@staticmethod
|
||||
def log_request():
|
||||
if request.endpoint and "static" not in request.endpoint:
|
||||
user = session.get("user", "Anonymous")
|
||||
ip = request.remote_addr
|
||||
|
||||
current_app.logger.info(
|
||||
f"{request.method} {request.path}",
|
||||
extra={"user": user, "ip": ip}
|
||||
)
|
||||
|
||||
# ---------------------------------------
|
||||
# Custom Action Logging
|
||||
# ---------------------------------------
|
||||
@staticmethod
|
||||
def log_action(action, details=""):
|
||||
"""Log user actions with timestamp, user, action, and details."""
|
||||
logData = LogData()
|
||||
logData.WriteLog(action, details="")
|
||||
|
||||
|
||||
class LogData:
|
||||
filepath = ""
|
||||
timestamp = None
|
||||
|
||||
def __init__(self):
|
||||
self.filepath = os.path.join(current_app.root_path, 'activity.log')
|
||||
self.timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
user = session.get("user", "Anonymous")
|
||||
ip = request.remote_addr
|
||||
current_app.logger.info(f"{action} | {details}",extra={"user": user, "ip": ip})
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from flask import Blueprint, render_template, request, redirect, url_for, flash, session
|
||||
import os
|
||||
from functools import wraps
|
||||
from ldap3 import Server, Connection, ALL
|
||||
from ldap3.core.exceptions import LDAPException
|
||||
@@ -9,76 +10,51 @@ class LoginAuth:
|
||||
# Create Blueprint
|
||||
self.bp = Blueprint("auth", __name__)
|
||||
|
||||
# -------------------------------
|
||||
# LDAP CONFIGURATION
|
||||
# -------------------------------
|
||||
self.LDAP_SERVER = "ldap://localhost:389"
|
||||
|
||||
self.BASE_DN = "ou=users,dc=lcepl,dc=org" # LDAP Users DN
|
||||
# LDAP CONFIG
|
||||
self.LDAP_SERVER = os.getenv("LDAP_SERVER", "ldap://host.docker.internal:389")
|
||||
self.BASE_DN = "ou=users,dc=lcepl,dc=org"
|
||||
|
||||
# -------------------------------
|
||||
# LOGIN ROUTE
|
||||
# -------------------------------
|
||||
# @self.bp.route('/login', methods=['GET', 'POST'])
|
||||
# def login():
|
||||
# if request.method == 'POST':
|
||||
# username = request.form.get("username")
|
||||
# password = request.form.get("password")
|
||||
# if not username or not password:
|
||||
# flash("Username and password are required!", "danger")
|
||||
# return render_template("login.html")
|
||||
# user_dn = f"uid={username},{self.BASE_DN}"
|
||||
# server = Server(self.LDAP_SERVER, get_info=ALL)
|
||||
# try:
|
||||
# # Attempt LDAP bind
|
||||
# conn = Connection(server, user=user_dn, password=password, auto_bind=True)
|
||||
# if conn.bound:
|
||||
# session['user'] = username
|
||||
# flash(f"Login successful! Welcome {username}", "success")
|
||||
# return redirect(url_for('welcome'))
|
||||
# else:
|
||||
# flash("Invalid username or password!", "danger")
|
||||
# except LDAPException as e:
|
||||
# flash(f"LDAP login failed: {str(e)}", "danger")
|
||||
# finally:
|
||||
# if 'conn' in locals():
|
||||
# conn.unbind()
|
||||
# # GET request: show login form
|
||||
# return render_template("login.html")
|
||||
# Register Routes
|
||||
self.bp.add_url_rule("/login", view_func=self.login, methods=["GET", "POST"])
|
||||
self.bp.add_url_rule("/logout", view_func=self.logout)
|
||||
|
||||
# ================= LOGIN =================
|
||||
def login(self):
|
||||
if request.method == "POST":
|
||||
username = request.form.get("username")
|
||||
password = request.form.get("password")
|
||||
|
||||
# LOGIN ROUTE
|
||||
@self.bp.route('/login', methods=['GET', 'POST'])
|
||||
def login():
|
||||
if request.method == 'POST':
|
||||
username = request.form.get("username")
|
||||
password = request.form.get("password")
|
||||
# Dummy validation — REPLACE with DB check later
|
||||
if username == "admin" and password == "admin123":
|
||||
session['user'] = username
|
||||
flash("Login successful!", "success")
|
||||
return redirect(url_for('welcome'))
|
||||
if not username or not password:
|
||||
flash("Username and password are required!", "danger")
|
||||
return render_template("login.html")
|
||||
|
||||
user_dn = f"uid={username},{self.BASE_DN}"
|
||||
server = Server(self.LDAP_SERVER, get_info=ALL)
|
||||
|
||||
try:
|
||||
conn = Connection(server, user=user_dn, password=password, auto_bind=True)
|
||||
|
||||
if conn.bound:
|
||||
session["user"] = username
|
||||
flash(f"Login successful! Welcome {username}", "success")
|
||||
conn.unbind()
|
||||
return redirect(url_for("welcome"))
|
||||
else:
|
||||
flash("Invalid username or password!", "danger")
|
||||
return render_template("login.html")
|
||||
|
||||
|
||||
# -------------------------------
|
||||
# LOGOUT ROUTE
|
||||
# -------------------------------
|
||||
@self.bp.route('/logout')
|
||||
def logout():
|
||||
session.clear()
|
||||
flash("Logged out successfully!", "success")
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
# ===================================================
|
||||
# LOGIN REQUIRED DECORATOR INSIDE CLASS
|
||||
# ===================================================
|
||||
except LDAPException as e:
|
||||
flash(f"LDAP login failed: {str(e)}", "danger")
|
||||
|
||||
return render_template("login.html")
|
||||
|
||||
# ================= LOGOUT =================
|
||||
def logout(self):
|
||||
session.clear()
|
||||
flash("Logged out successfully!", "success")
|
||||
return redirect(url_for("auth.login"))
|
||||
|
||||
# ================= LOGIN REQUIRED =================
|
||||
def login_required(self, f):
|
||||
"""
|
||||
Protect routes: redirect to login if user not authenticated.
|
||||
"""
|
||||
@wraps(f)
|
||||
def wrapper(*args, **kwargs):
|
||||
if "user" not in session:
|
||||
|
||||
@@ -13,7 +13,7 @@ class MatCreditHandler:
|
||||
def fetch_all(self):
|
||||
try:
|
||||
|
||||
self.cursor.callproc("GetMatCedit")
|
||||
self.cursor.callproc("GetMatCredit")
|
||||
result_sets = self.cursor.stored_results()
|
||||
mat_rows = next(result_sets).fetchall()
|
||||
utilization_rows = next(result_sets).fetchall()
|
||||
|
||||
Reference in New Issue
Block a user