security: audit complet + patch vulnerabilitati critice v1.3.5
Audit de securitate executat pe Backend (FastAPI) si Frontend (Next.js/Dexie). 12 vulnerabilitati identificate (4 CRITICE, 4 HIGH, 3 MEDIUM, 1 LOW). Patch-uri aplicate direct: - [C-02] Eliminat bypass autentificare pentru useri fara parola (users.py) - [C-03] Parola default Admin inlocuita cu secrets.token_urlsafe(16) (users.py) - [H-01] LDAP injection fix: escape_filter_chars pe username (users.py) - [H-03] Validare MIME + limita 10MB pe /items/extract-label (items.py) - [M-01] CORS fix: allow_origins din env ALLOWED_ORIGINS, nu wildcard (main.py) - [M-03] Toate print() LDAP inlocuite cu log.debug() (users.py) Raport complet: dev_docs/SECURITY_REPORT.md Actiuni arhitecturale ramase (JWT enforcement, rate limiting): SESSION_STATE.md Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Daniel Bedeleanu
@@ -37,16 +37,29 @@ def read_item(item_id: int, db: Session = Depends(get_db)):
|
||||
raise HTTPException(status_code=404, detail="Item not found")
|
||||
return item
|
||||
|
||||
_ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"}
|
||||
_MAX_IMAGE_SIZE = 10 * 1024 * 1024 # 10 MB
|
||||
|
||||
@router.post("/extract-label")
|
||||
async def extract_label(file: UploadFile = File(...)):
|
||||
from ..ai_vision import extract_label_info
|
||||
|
||||
# Read image content
|
||||
|
||||
# [SECURITY FIX H-03] Validare tip MIME și dimensiune maximă
|
||||
if file.content_type not in _ALLOWED_IMAGE_TYPES:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
|
||||
detail=f"Tip fișier nepermis: {file.content_type}. Acceptat: {', '.join(_ALLOWED_IMAGE_TYPES)}"
|
||||
)
|
||||
|
||||
contents = await file.read()
|
||||
|
||||
# Process with Gemini
|
||||
|
||||
if len(contents) > _MAX_IMAGE_SIZE:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
||||
detail="Fișierul depășește limita de 10MB."
|
||||
)
|
||||
|
||||
result = extract_label_info(contents)
|
||||
|
||||
return result
|
||||
|
||||
@router.post("/", response_model=schemas.Item, status_code=status.HTTP_201_CREATED)
|
||||
|
||||
@@ -1,11 +1,14 @@
|
||||
import secrets
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
from typing import List
|
||||
from passlib.context import CryptContext
|
||||
import ldap3
|
||||
from ldap3.utils.conv import escape_filter_chars
|
||||
import json
|
||||
import os
|
||||
from .. import models, schemas, database
|
||||
from ..logger import log
|
||||
|
||||
router = APIRouter(prefix="/users", tags=["users"])
|
||||
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
||||
@@ -25,22 +28,24 @@ def authenticate_ldap(username, password):
|
||||
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}")
|
||||
|
||||
log.debug(f"LDAP: Attempting bind for DN: {user_dn}")
|
||||
|
||||
conn = ldap3.Connection(server, user=user_dn, password=password, auto_bind=True)
|
||||
print(f"DEBUG LDAP: Bind successful for {user_dn}")
|
||||
|
||||
log.debug(f"LDAP: Bind successful for {user_dn}")
|
||||
|
||||
# Search for the user to get their CANONICAL DN
|
||||
# [SECURITY FIX H-01] Escape username before interpolating into LDAP filter
|
||||
base_dn = config.get("base_dn", "dc=example,dc=org")
|
||||
search_filter = f"(|(cn={username})(uid={username}))"
|
||||
safe_username = escape_filter_chars(username)
|
||||
search_filter = f"(|(cn={safe_username})(uid={safe_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.")
|
||||
log.debug(f"LDAP: User 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}")
|
||||
log.debug(f"LDAP: Canonical DN found: {real_user_dn}")
|
||||
|
||||
# Check roles based on group membership
|
||||
assigned_role = None
|
||||
@@ -67,14 +72,14 @@ def authenticate_ldap(username, password):
|
||||
else:
|
||||
full_group_dn = group_name
|
||||
|
||||
print(f"DEBUG LDAP: Checking membership in group: {full_group_dn}")
|
||||
log.debug(f"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}")
|
||||
log.debug(f"LDAP: User is in group {group_name}, assigning role: {target_role}")
|
||||
potential_roles.append(target_role)
|
||||
|
||||
if "admin" in potential_roles:
|
||||
@@ -86,7 +91,7 @@ def authenticate_ldap(username, password):
|
||||
|
||||
return assigned_role
|
||||
except Exception as e:
|
||||
print(f"DEBUG LDAP: Auth Error: {str(e)}")
|
||||
log.debug(f"LDAP: Auth Error: {str(e)}")
|
||||
return None
|
||||
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
||||
|
||||
@@ -109,15 +114,18 @@ def get_users(db: Session = Depends(get_db)):
|
||||
users = db.query(models.User).all()
|
||||
# Auto-seed if empty
|
||||
if not users:
|
||||
# [SECURITY FIX C-03] Generare parolă aleatoare în loc de "admin" hardcodat
|
||||
initial_password = secrets.token_urlsafe(16)
|
||||
new_user = models.User(
|
||||
username="Admin",
|
||||
role="admin",
|
||||
username="Admin",
|
||||
role="admin",
|
||||
origin="local",
|
||||
hashed_password=get_password_hash("admin") # Default password
|
||||
hashed_password=get_password_hash(initial_password)
|
||||
)
|
||||
db.add(new_user)
|
||||
db.commit()
|
||||
db.refresh(new_user)
|
||||
log.warning(f"[SECURITY] Admin initial seeded. Temporary password: {initial_password} — SCHIMBAȚI IMEDIAT!")
|
||||
return [new_user]
|
||||
return users
|
||||
|
||||
@@ -144,8 +152,9 @@ def login(form_data: schemas.UserLogin, db: Session = Depends(get_db)):
|
||||
if verify_password(form_data.password, user.hashed_password):
|
||||
authenticated = True
|
||||
elif user and not user.hashed_password:
|
||||
# Legacy user without password - allow skip for now or force set
|
||||
authenticated = True
|
||||
# [SECURITY FIX C-02] Bypass-ul pentru utilizatori fără parolă a fost eliminat.
|
||||
# Utilizatorii LDAP trebuie să se autentifice prin fluxul LDAP de mai jos.
|
||||
pass
|
||||
|
||||
# If local failed, try LDAP
|
||||
if not authenticated:
|
||||
@@ -231,7 +240,7 @@ def test_ldap_connection(config: dict):
|
||||
port = 3890
|
||||
|
||||
# Try raw socket first
|
||||
print(f"DEBUG: Probing raw socket {host}:{port}")
|
||||
log.debug(f"LDAP test: Probing raw socket {host}:{port}")
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
s.settimeout(5)
|
||||
result = s.connect_ex((host, port))
|
||||
|
||||
Reference in New Issue
Block a user