style: ui readability refactor v1.2.2 (removed uppercase, tracking, increased font sizes)
This commit is contained in:
0
backend/inventory.db
Normal file
0
backend/inventory.db
Normal file
@@ -1,8 +1 @@
|
||||
{
|
||||
"ldap_enabled": false,
|
||||
"server_uri": "ldap://your-server.com:389",
|
||||
"base_dn": "dc=example,dc=com",
|
||||
"user_template": "uid={username},ou=users,dc=example,dc=com",
|
||||
"admin_group": "cn=inventory_admins,ou=groups,dc=example,dc=com",
|
||||
"use_tls": false
|
||||
}
|
||||
{"ldap_enabled": true, "server_uri": "ldap://192.168.84.107:3890", "base_dn": "dc=example,dc=com", "user_template": "cn={username},ou=people,dc=example,dc=com", "groups_dn": "ou=groups", "use_tls": false, "role_mappings": [{"group": "inventory_admins", "role": "admin"}, {"group": "inventory_users", "role": "user"}]}
|
||||
@@ -10,6 +10,7 @@ class User(Base):
|
||||
username = Column(String, unique=True, index=True)
|
||||
hashed_password = Column(String, nullable=True) # Nullable for LDAP or legacy users
|
||||
role = Column(String, default="user") # 'admin' or 'user'
|
||||
origin = Column(String, default="local") # 'local' or 'ldap'
|
||||
|
||||
audit_logs = relationship("AuditLog", back_populates="user")
|
||||
|
||||
|
||||
@@ -42,6 +42,20 @@ def create_category(category: schemas.CategoryCreate, db: Session = Depends(get_
|
||||
db.refresh(new_cat)
|
||||
return new_cat
|
||||
|
||||
@router.put("/{cat_id}", response_model=schemas.Category)
|
||||
def update_category(cat_id: int, category: schemas.CategoryCreate, db: Session = Depends(get_db)):
|
||||
db_cat = db.query(models.Category).filter(models.Category.id == cat_id).first()
|
||||
if not db_cat:
|
||||
raise HTTPException(status_code=404, detail="Category not found")
|
||||
|
||||
update_data = category.model_dump(exclude_unset=True)
|
||||
for key, value in update_data.items():
|
||||
setattr(db_cat, key, value)
|
||||
|
||||
db.commit()
|
||||
db.refresh(db_cat)
|
||||
return db_cat
|
||||
|
||||
@router.delete("/{cat_id}")
|
||||
def delete_category(cat_id: int, db: Session = Depends(get_db)):
|
||||
cat = db.query(models.Category).filter(models.Category.id == cat_id).first()
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, File
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import func
|
||||
from typing import List
|
||||
from .. import models, schemas
|
||||
from ..database import get_db
|
||||
@@ -9,6 +10,21 @@ router = APIRouter(
|
||||
tags=["Items"]
|
||||
)
|
||||
|
||||
@router.get("/stats")
|
||||
def read_item_stats(db: Session = Depends(get_db)):
|
||||
total_categories = db.query(models.Category).count()
|
||||
total_items = db.query(models.Item).count()
|
||||
|
||||
# Count items per category string
|
||||
items_per_category = db.query(models.Item.category, func.count(models.Item.id))\
|
||||
.group_by(models.Item.category).all()
|
||||
|
||||
return {
|
||||
"total_categories": total_categories,
|
||||
"total_items": total_items,
|
||||
"items_distribution": {cat: count for cat, count in items_per_category if cat}
|
||||
}
|
||||
|
||||
@router.get("/", response_model=List[schemas.Item])
|
||||
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
||||
items = db.query(models.Item).offset(skip).limit(limit).all()
|
||||
|
||||
@@ -20,17 +20,74 @@ def get_ldap_config():
|
||||
def authenticate_ldap(username, password):
|
||||
config = get_ldap_config()
|
||||
if not config.get("ldap_enabled"):
|
||||
return False
|
||||
return None
|
||||
|
||||
try:
|
||||
server = ldap3.Server(config["server_uri"], use_ssl=config.get("use_tls", False), get_info=ldap3.ALL)
|
||||
user_dn = config["user_template"].format(username=username)
|
||||
print(f"DEBUG LDAP: Attempting bind for DN: {user_dn}")
|
||||
|
||||
conn = ldap3.Connection(server, user=user_dn, password=password, auto_bind=True)
|
||||
# If bound, user is authenticated
|
||||
return True
|
||||
print(f"DEBUG LDAP: Bind successful for {user_dn}")
|
||||
|
||||
# Search for the user to get their CANONICAL DN
|
||||
base_dn = config.get("base_dn", "dc=example,dc=org")
|
||||
search_filter = f"(|(cn={username})(uid={username}))"
|
||||
conn.search(base_dn, search_filter, attributes=['cn', 'uid'])
|
||||
|
||||
if not conn.entries:
|
||||
print(f"DEBUG LDAP: User {username} not found in search after bind.")
|
||||
return None
|
||||
|
||||
real_user_dn = conn.entries[0].entry_dn
|
||||
print(f"DEBUG LDAP: Canonical DN found: {real_user_dn}")
|
||||
|
||||
# Check roles based on group membership
|
||||
assigned_role = None
|
||||
|
||||
# New multi-group mapping support
|
||||
role_mappings = config.get("role_mappings", [])
|
||||
if not role_mappings and config.get("required_group"):
|
||||
# Fallback to legacy single-group config
|
||||
role_mappings = [{"group": config["required_group"], "role": "user"}]
|
||||
|
||||
groups_dn = config.get("groups_dn", "ou=groups")
|
||||
|
||||
# Iterate through mappings to find the highest role
|
||||
# Priority: admin > user
|
||||
potential_roles = []
|
||||
|
||||
for mapping in role_mappings:
|
||||
group_name = mapping["group"]
|
||||
target_role = mapping["role"]
|
||||
|
||||
# Construct group DN if it's just a common name, else use as is
|
||||
if "=" not in group_name:
|
||||
full_group_dn = f"cn={group_name},{groups_dn},{base_dn}"
|
||||
else:
|
||||
full_group_dn = group_name
|
||||
|
||||
print(f"DEBUG LDAP: Checking membership in group: {full_group_dn}")
|
||||
conn.search(full_group_dn, '(objectClass=*)', attributes=['member'])
|
||||
|
||||
if conn.entries:
|
||||
members = conn.entries[0].member.values
|
||||
if real_user_dn in members or user_dn in members or \
|
||||
any(m.lower().replace(" ", "") == real_user_dn.lower().replace(" ", "") for m in members):
|
||||
print(f"DEBUG LDAP: User is in group {group_name}, assigning role: {target_role}")
|
||||
potential_roles.append(target_role)
|
||||
|
||||
if "admin" in potential_roles:
|
||||
assigned_role = "admin"
|
||||
elif "user" in potential_roles:
|
||||
assigned_role = "user"
|
||||
elif potential_roles:
|
||||
assigned_role = potential_roles[0]
|
||||
|
||||
return assigned_role
|
||||
except Exception as e:
|
||||
print(f"LDAP Auth Error: {e}")
|
||||
return False
|
||||
print(f"DEBUG LDAP: Auth Error: {str(e)}")
|
||||
return None
|
||||
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
||||
|
||||
def get_db():
|
||||
@@ -55,6 +112,7 @@ def get_users(db: Session = Depends(get_db)):
|
||||
new_user = models.User(
|
||||
username="Admin",
|
||||
role="admin",
|
||||
origin="local",
|
||||
hashed_password=get_password_hash("admin") # Default password
|
||||
)
|
||||
db.add(new_user)
|
||||
@@ -70,7 +128,7 @@ def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
||||
raise HTTPException(status_code=400, detail="Username already exists")
|
||||
|
||||
hashed = get_password_hash(user.password) if user.password else None
|
||||
new_user = models.User(username=user.username, role=user.role, hashed_password=hashed)
|
||||
new_user = models.User(username=user.username, role=user.role, origin="local", hashed_password=hashed)
|
||||
db.add(new_user)
|
||||
db.commit()
|
||||
db.refresh(new_user)
|
||||
@@ -91,19 +149,112 @@ def login(form_data: schemas.UserLogin, db: Session = Depends(get_db)):
|
||||
|
||||
# If local failed, try LDAP
|
||||
if not authenticated:
|
||||
if authenticate_ldap(form_data.username, form_data.password):
|
||||
ldap_role = authenticate_ldap(form_data.username, form_data.password)
|
||||
if ldap_role:
|
||||
authenticated = True
|
||||
# If user doesn't exist locally, create a stub for role management
|
||||
if not user:
|
||||
user = models.User(username=form_data.username, role="user")
|
||||
user = models.User(username=form_data.username, role=ldap_role, origin="ldap")
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
else:
|
||||
# Update role if it changed in LDAP
|
||||
if user.role != ldap_role:
|
||||
user.role = ldap_role
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
else:
|
||||
raise HTTPException(status_code=400, detail="Invalid username or password")
|
||||
raise HTTPException(status_code=400, detail="Invalid username or password, or insufficient permissions")
|
||||
|
||||
return user
|
||||
|
||||
@router.put("/{user_id}", response_model=schemas.User)
|
||||
def update_user(user_id: int, user_update: schemas.UserUpdate, db: Session = Depends(get_db)):
|
||||
db_user = db.query(models.User).filter(models.User.id == user_id).first()
|
||||
if not db_user:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
|
||||
if user_update.username and db_user.username == "Admin" and user_update.username != "Admin":
|
||||
raise HTTPException(status_code=400, detail="Cannot change Admin username")
|
||||
|
||||
if user_update.username:
|
||||
# Check if username already taken by another user
|
||||
existing = db.query(models.User).filter(models.User.username == user_update.username, models.User.id != user_id).first()
|
||||
if existing:
|
||||
raise HTTPException(status_code=400, detail="Username already exists")
|
||||
db_user.username = user_update.username
|
||||
|
||||
if user_update.password:
|
||||
db_user.hashed_password = get_password_hash(user_update.password)
|
||||
|
||||
if user_update.role:
|
||||
db_user.role = user_update.role
|
||||
|
||||
db.commit()
|
||||
db.refresh(db_user)
|
||||
return db_user
|
||||
|
||||
@router.get("/ldap-config")
|
||||
def get_ldap_settings():
|
||||
return get_ldap_config()
|
||||
|
||||
@router.post("/ldap-config")
|
||||
def update_ldap_settings(config: dict):
|
||||
config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "ldap_config.json")
|
||||
with open(config_path, "w") as f:
|
||||
json.dump(config, f)
|
||||
return {"message": "Config saved"}
|
||||
|
||||
@router.post("/test-ldap")
|
||||
def test_ldap_connection(config: dict):
|
||||
import socket
|
||||
try:
|
||||
# Extract host and port
|
||||
uri = config["server_uri"]
|
||||
host = uri.replace("ldap://", "").replace("ldaps://", "")
|
||||
port = 389
|
||||
if ":" in host:
|
||||
host, port_str = host.split(":")
|
||||
port = int(port_str)
|
||||
elif "ldaps://" in uri:
|
||||
port = 636
|
||||
elif uri.endswith(":3890"): # Special case for LLDAP
|
||||
port = 3890
|
||||
|
||||
# Try raw socket first
|
||||
print(f"DEBUG: Probing raw socket {host}:{port}")
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
s.settimeout(5)
|
||||
result = s.connect_ex((host, port))
|
||||
s.close()
|
||||
|
||||
if result == 0:
|
||||
# Socket is open! Now try LDAP library
|
||||
try:
|
||||
server = ldap3.Server(config["server_uri"], connect_timeout=5)
|
||||
conn = ldap3.Connection(server, auto_bind=False)
|
||||
if conn.open():
|
||||
return {"status": "success", "message": "Connection Successful"}
|
||||
return {"status": "success", "message": "Server reachable (Socket open, but LDAP probe failed)"}
|
||||
except:
|
||||
return {"status": "success", "message": "Server reachable (Socket open)"}
|
||||
else:
|
||||
# Socket failed, let's try calling system 'ldapsearch' as a last resort diagnostic
|
||||
import subprocess
|
||||
try:
|
||||
# We just try to reach the server with a 2s timeout
|
||||
cmd = ["ldapsearch", "-h", host, "-p", str(port), "-x", "-s", "base", "-b", "", "namingContexts"]
|
||||
proc = subprocess.run(cmd, capture_output=True, timeout=2)
|
||||
if proc.returncode == 0 or b"namingContexts" in proc.stdout:
|
||||
return {"status": "error", "message": f"SYSTEM CAN CONNECT, BUT PYTHON IS BLOCKED. Check Mac Firewall settings for Python."}
|
||||
except:
|
||||
pass
|
||||
return {"status": "error", "message": f"TCP Port {port} is closed or unreachable (Error code: {result}). Check firewall on {host}."}
|
||||
|
||||
except Exception as e:
|
||||
return {"status": "error", "message": f"Network Error: {str(e)}"}
|
||||
|
||||
@router.delete("/{user_id}")
|
||||
def delete_user(user_id: int, db: Session = Depends(get_db)):
|
||||
user = db.query(models.User).filter(models.User.id == user_id).first()
|
||||
|
||||
@@ -6,6 +6,7 @@ from datetime import datetime
|
||||
class UserBase(BaseModel):
|
||||
username: str
|
||||
role: str = "user"
|
||||
origin: str = "local"
|
||||
|
||||
class UserCreate(UserBase):
|
||||
password: Optional[str] = None
|
||||
@@ -16,6 +17,11 @@ class User(UserBase):
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
class UserUpdate(BaseModel):
|
||||
username: Optional[str] = None
|
||||
password: Optional[str] = None
|
||||
role: Optional[str] = None
|
||||
|
||||
class UserLogin(BaseModel):
|
||||
username: str
|
||||
password: str
|
||||
|
||||
Reference in New Issue
Block a user