bede/tfm_ainventory
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Build [v1.5.0] - Box Management & Label Printing

Browse Source
This commit is contained in:
Daniel Bedeleanu
2026-04-12 21:30:50 +03:00
parent e383b97d44
commit 2e5c666cc8
18 changed files with 1007 additions and 336 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
backend/routers/items.py
View File

@@ -2,6 +2,7 @@ from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, File,
from sqlalchemy.orm import Session
from sqlalchemy import func
from typing import List
import json
from slowapi import Limiter
from slowapi.util import get_remote_address
from .. import models, schemas, auth
@@ -106,10 +107,27 @@ def create_item(
db.refresh(db_item)
# Audit log the creation — [M-02] user_id from token, not from body
# Capture full snapshot
item_snapshot = {
"barcode": db_item.barcode,
"name": db_item.name,
"category": db_item.category,
"type": db_item.type,
"part_number": db_item.part_number,
"color": db_item.color,
"specs": db_item.specs,
"box_label": db_item.box_label,
"image_url": db_item.image_url
}
audit = models.AuditLog(
user_id=current_user.sub,
action="CREATE_ITEM",
target_item_id=db_item.id,
target_item_name=db_item.name,
target_item_pn=db_item.part_number,
target_item_barcode=db_item.barcode,
target_snapshot=json.dumps(item_snapshot),
quantity_change=item.quantity
)
db.add(audit)
@@ -148,15 +166,39 @@ def delete_item(
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
# [AUDIT] Log the deletion to disk file via logger.py
# [AUDIT] Log the deletion to database and disk
from ..logger import log
log.warning(f"USER[{current_user.sub}] DELETING ITEM: ID={item_id}, Name={db_item.name}, PN={db_item.part_number}")
item_snapshot = {
"barcode": db_item.barcode,
"name": db_item.name,
"category": db_item.category,
"type": db_item.type,
"part_number": db_item.part_number,
"color": db_item.color,
"specs": db_item.specs,
"box_label": db_item.box_label,
"image_url": db_item.image_url,
"final_quantity": db_item.quantity
}
audit = models.AuditLog(
user_id=current_user.sub,
action="DELETE_ITEM",
target_item_id=item_id,
target_item_name=db_item.name,
target_item_pn=db_item.part_number,
target_item_barcode=db_item.barcode,
target_snapshot=json.dumps(item_snapshot),
details=f"Final Quantity: {db_item.quantity}"
)
db.add(audit)
# [CLEANUP] Delete related InterventionItems to prevent foreign key issues
db.query(models.InterventionItem).filter(models.InterventionItem.item_id == item_id).delete()
# Audit Logs in database are NOT deleted here to preserve history of actions
db.delete(db_item)
db.commit()
return {"message": "Item deleted successfully. History logs preserved."}

64
backend/routers/operations.py
View File

@@ -3,6 +3,7 @@ from typing import List
from sqlalchemy.orm import Session
from .. import models, schemas, auth
from ..database import get_db
import json
router = APIRouter(
prefix="/operations",
@@ -27,10 +28,21 @@ def check_in_item(
item.quantity += op.quantity
# Create Mandatory Audit Log — [M-02] user_id from token
item_snapshot = {
"barcode": item.barcode,
"name": item.name,
"category": item.category,
"part_number": item.part_number
}
audit = models.AuditLog(
user_id=current_user.sub,
action="CHECK_IN",
target_item_id=item.id,
target_item_name=item.name,
target_item_pn=item.part_number,
target_item_barcode=item.barcode,
target_snapshot=json.dumps(item_snapshot),
quantity_change=op.quantity
)
@@ -60,10 +72,21 @@ def check_out_item(
item.quantity -= op.quantity
# Create Mandatory Audit Log
item_snapshot = {
"barcode": item.barcode,
"name": item.name,
"category": item.category,
"part_number": item.part_number
}
audit = models.AuditLog(
user_id=current_user.sub,
action="CHECK_OUT",
target_item_id=item.id,
target_item_name=item.name,
target_item_pn=item.part_number,
target_item_barcode=item.barcode,
target_snapshot=json.dumps(item_snapshot),
quantity_change=-op.quantity
)
@@ -93,10 +116,21 @@ def trash_item(
item.quantity -= op.quantity
# Create Mandatory Audit Log with TRASH action and reason in details
item_snapshot = {
"barcode": item.barcode,
"name": item.name,
"category": item.category,
"part_number": item.part_number
}
audit = models.AuditLog(
user_id=current_user.sub,
action="TRASH",
target_item_id=item.id,
target_item_name=item.name,
target_item_pn=item.part_number,
target_item_barcode=item.barcode,
target_snapshot=json.dumps(item_snapshot),
quantity_change=-op.quantity,
details=op.reason
)
@@ -130,10 +164,21 @@ def bulk_check_out(
item.quantity -= op.quantity
# Log individual audit for this item
item_snapshot = {
"barcode": item.barcode,
"name": item.name,
"category": item.category,
"part_number": item.part_number
}
audit = models.AuditLog(
user_id=current_user.sub,
action="BULK_CHECK_OUT",
target_item_id=item.id,
target_item_name=item.name,
target_item_pn=item.part_number,
target_item_barcode=item.barcode,
target_snapshot=json.dumps(item_snapshot),
quantity_change=-op.quantity
)
db.add(audit)
@@ -182,10 +227,21 @@ def bulk_sync(
continue
# Log audit with original offline timestamp and UUID
item_snapshot = {
"barcode": item.barcode,
"name": item.name,
"category": item.category,
"part_number": item.part_number
}
audit = models.AuditLog(
user_id=current_user.sub,
action=op.type,
target_item_id=item.id,
target_item_name=item.name,
target_item_pn=item.part_number,
target_item_barcode=item.barcode,
target_snapshot=json.dumps(item_snapshot),
quantity_change=change,
timestamp=op.timestamp,
uuid=op.uuid,
@@ -215,6 +271,10 @@ def get_logs(
models.User.username,
models.AuditLog.action,
models.AuditLog.target_item_id,
models.AuditLog.target_item_name,
models.AuditLog.target_item_pn,
models.AuditLog.target_item_barcode,
models.AuditLog.target_snapshot,
models.AuditLog.quantity_change,
models.AuditLog.details
).join(models.User, models.AuditLog.user_id == models.User.id).order_by(models.AuditLog.timestamp.desc()).limit(limit).all()
@@ -228,6 +288,10 @@ def get_logs(
"username": l.username,
"action": l.action,
"target_item_id": l.target_item_id,
"target_item_name": l.target_item_name,
"target_item_pn": l.target_item_pn,
"target_item_barcode": l.target_item_barcode,
"target_snapshot": l.target_snapshot,
"quantity_change": l.quantity_change,
"details": l.details
} for l in logs_with_users

67
backend/routers/users.py
View File

@@ -6,8 +6,10 @@ from slowapi import Limiter
from slowapi.util import get_remote_address
from passlib.context import CryptContext
import ldap3
from ldap3 import Tls
from ldap3.utils.conv import escape_filter_chars
from ldap3.utils.dn import escape_rdn
import ssl
import json
import os
from .. import models, schemas, database, auth
@@ -34,7 +36,22 @@ def authenticate_ldap(username, password):
log.debug(f"LDAP: Config loaded: server_uri={config.get('server_uri')}, base_dn={config.get('base_dn')}")
try:
server = ldap3.Server(config["server_uri"], use_ssl=config.get("use_tls", False), get_info=ldap3.ALL)
tls_config = None
if config.get("use_tls", False):
if config.get("ignore_cert", False):
# [SECURITY] CERT_NONE is only for internal test environments with self-signed certs
tls_config = Tls(validate=ssl.CERT_NONE, version=ssl.PROTOCOL_TLSv1_2)
log.warning("LDAP: TLS Certificate Validation DISABLED (ignore_cert=true)")
else:
tls_config = Tls(validate=ssl.CERT_REQUIRED, version=ssl.PROTOCOL_TLSv1_2)
log.debug("LDAP: TLS Certificate Validation ENABLED (CERT_REQUIRED)")
server = ldap3.Server(
config["server_uri"],
use_ssl=config.get("use_tls", False),
tls=tls_config,
get_info=ldap3.ALL
)
log.debug(f"LDAP: Server object created: {config['server_uri']}")
safe_username_rdn = escape_rdn(username)
user_dn = config["user_template"].format(username=safe_username_rdn)
@@ -101,10 +118,31 @@ def authenticate_ldap(username, password):
return assigned_role
except Exception as e:
log.error(f"LDAP: Auth Error: {type(e).__name__}: {str(e)}")
err_msg = str(e)
err_type = type(e).__name__
log.error(f"LDAP: Auth Error: {err_type}: {err_msg}")
# Broad detection for SSL/TLS certificate/handshake or connectivity errors
# handles both ldapsearch style "Can't contact" and ldap3 style "socket ssl wrapping error"
ssl_indicators = ["certificate", "ssl", "tls", "handshake", "verify failed", "contact", "socket"]
if any(ind in err_msg.lower() for ind in ssl_indicators):
log.warning(f"LDAP: SSL/TLS or Connectivity issue detected: {err_msg}")
# User-friendly error message, hiding raw socket traces
friendly_msg = "Secure Connection Failed: The enterprise server's security certificate is not trusted or the connection dropped."
if config.get("use_tls"):
friendly_msg += " If this is an internal test environment, please ask an Admin to enable 'Ignore Certificate Validation'."
raise HTTPException(
status_code=401,
detail=friendly_msg
)
import traceback
log.debug(f"LDAP: Full traceback: {traceback.format_exc()}")
return None
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
def get_db():
@@ -313,7 +351,20 @@ def test_ldap_connection(
if result == 0:
# Socket is open! Now try LDAP library probe
try:
server = ldap3.Server(config["server_uri"], connect_timeout=5, get_info=ldap3.BASIC)
tls_config = None
if config.get("use_tls", False):
if config.get("ignore_cert", False):
tls_config = Tls(validate=ssl.CERT_NONE, version=ssl.PROTOCOL_TLSv1_2)
else:
tls_config = Tls(validate=ssl.CERT_REQUIRED, version=ssl.PROTOCOL_TLSv1_2)
server = ldap3.Server(
config["server_uri"],
connect_timeout=5,
get_info=ldap3.BASIC,
use_ssl=config.get("use_tls", False),
tls=tls_config
)
# Try a connection without auto-bind first to see if it's an LDAP server
conn = ldap3.Connection(server, auto_bind=False)
if conn.open():
@@ -324,7 +375,10 @@ def test_ldap_connection(
return {"status": "success", "message": "Connection Successful (Network reachable, protocol handshake restricted by server security)"}
except Exception as e:
# Any LDAP level error while socket is open is still a partial success
return {"status": "success", "message": f"Partial Success: TCP Port {port} is open, but LDAP handshake was rejected."}
err_msg = str(e)
if "certificate verify failed" in err_msg.lower() or "self signed certificate" in err_msg.lower():
return {"status": "error", "message": f"SSL/TLS Certificate Rejected: The server certificate is self-signed or invalid. Enable 'Ignore Certificate Validation' to bypass."}
return {"status": "success", "message": f"Partial Success: TCP Port {port} is open, but LDAP handshake was rejected: {err_msg}"}
else:
# Socket failed, let's try calling system 'ldapsearch' as a last resort diagnostic
import subprocess
@@ -355,6 +409,9 @@ def delete_user(
if user.username == "Admin":
raise HTTPException(status_code=400, detail="Cannot delete default Admin")
is_ldap = user.origin == "ldap"
db.delete(user)
db.commit()
return {"message": "User deleted"}
log.info(f"User {user_id} ({user.username}) deleted by Admin. Source: {user.origin}")
return {"message": "User deleted" if not is_ldap else "LDAP cache cleared for this user"}