Zurück zur Schwachstellen-Datenbank

Datenzugriffsoperationen außerhalb der erwarteten Datenmanager-Komponente

Beschreibung

Datenzugriffsoperationen außerhalb der erwarteten Datenmanager-Komponente treten auf, wenn ein Produkt eine dedizierte, zentrale Datenmanager-Komponente (wie ein Repository, DAO oder Data Access Layer) als Teil seines Designs verwendet, aber Code enthält, der Datenzugriffsoperationen durchführt, indem er diesen Manager umgeht. Anstatt die designierte Datenzugriffskomponente zu verwenden, greift Code in anderen Schichten direkt auf die Datenbank, das Dateisystem oder andere Datenspeicher zu. Dies erzeugt inkonsistente Datenzugriffsmuster und umgeht zentralisierte Belange wie Caching, Logging, Validierung und Sicherheitsprüfungen.

Risiko

Obwohl hauptsächlich ein architektonisches Qualitätsproblem, hat das Umgehen des Datenmanagers Sicherheitsimplikationen. Zentralisierte Sicherheitsprüfungen (Autorisierung, Eingabevalidierung, Audit-Logging), die im Datenmanager implementiert sind, werden übersprungen. Caching-Mechanismen werden umgangen, was möglicherweise Leistungsprobleme verursacht, die DoS-Angriffe ermöglichen. Inkonsistenter Datenzugriff erschwert Sicherheitsaudits. SQL-Injection oder andere Injection-Schwachstellen sind wahrscheinlicher, wenn Datenzugriffscode verstreut ist. Transaktionsmanagement kann inkonsistent sein, was zu Datenintegritätsproblemen führt. Connection Pooling kann unterlaufen werden, was Ressourcenerschöpfung verursacht.

Lösung

Erzwingen Sie Datenzugriff ausschließlich durch die designierte Datenmanager-Komponente. Verwenden Sie Dependency Injection, um Datenzugriffskomponenten bereitzustellen, anstatt direkte Instanziierung zu erlauben. Wenden Sie architektonische Fitness-Funktionen in CI/CD an, um direkten Datenzugriff außerhalb des Managers zu erkennen. Machen Sie Datenbankverbindungsdetails privat für die Datenzugriffsschicht. Verwenden Sie Code-Reviews, um Verstoße zu fangen. Erwägen Sie die Verwendung von Kompilierzeit- oder Laufzeit-Zugriffskontrollen, um direkten Datenspeicherzugriff zu verhindern. Dokumentieren Sie die Architektur klar, damit alle Entwickler das Muster verstehen.

Häufige Auswirkungen

AuswirkungDetails
AndereBereich: Ändere

Leistungsreduktion - Das Umgehen des optimierten zentralen Datenmanagers kann Leistungsverschlechterung und potenzielle Sicherheitsschwachstellen verursachen.
AndereBereich: Ändere

Reduzierte Wartbarkeit - Verstreuter Datenzugriffscode ist schwerer konsistent zu warten und zu sichern.
IntegritätBereich: Integrität

Sicherheitsumgehung - Sicherheitskontrollen, die im Datenmanager implementiert sind, werden umgangen.

Beispielcode

Anfälliger Code

// Architektur: Sollte Repository-Pattern für allen Datenzugriff verwenden

// Datenmanager (Repository) - die designierte Datenzugriffskomponente
@Repository
public class UserRepository {
    private final EntityManager em;

    public User findById(Long id) {
        // Zentralisierte Belange:
        // - Caching
        // - Audit-Logging
        // - Autorisierungsprüfung
        auditService.log("user.read", id);
        return em.find(User.class, id);
    }

    public void save(User user) {
        // Validierung, Audit usw.
        validateUser(user);
        auditService.log("user.save", user.getId());
        em.persist(user);
    }
}

// Anfällig: Service umgeht das Repository
@Service
public class VulnerableUserService {

    @Autowired
    private UserRepository userRepository;  // Hat das richtige Repository

    @PersistenceContext
    private EntityManager entityManager;  // Aber hat auch direkten EM-Zugriff!

    public User getUserFast(Long id) {
        // Anfällig: Umgeht Repository zur "Optimierung"
        // Überspringt: Caching, Audit-Logging, Autorisierung
        return entityManager.find(User.class, id);
    }

    public void updateUserDirect(Long id, String newEmail) {
        // Anfällig: Direkte Datenbankaktualisierung
        // Umgeht alle Repository-Validierung und -Logging
        Query query = entityManager.createQuery(
            "UPDATE User u SET u.email = :email WHERE u.id = :id");
        query.setParameter("email", newEmail);
        query.setParameter("id", id);
        query.executeUpdate();
    }
}

# Architektur: Datenmanager-Muster mit SqlAlchemy

# Datenmanager - zentralisierter Datenzugriff
class UserRepository:
    def __init__(self, session, audit_service):
        self._session = session
        self._audit = audit_service

    def get_by_id(self, user_id: int) -> User:
        # Zentralisiert: Audit, Cache-Prüfung, Autorisierung
        self._audit.log("user.read", user_id)
        return self._session.query(User).get(user_id)

    def save(self, user: User):
        # Validierung, Audit-Logging
        self._validate(user)
        self._audit.log("user.save", user.id)
        self._session.add(user)
        self._session.commit()


# Anfällig: Service umgeht Repository
class VulnerableUserService:
    def __init__(self, user_repo: UserRepository, db_session):
        self.user_repo = user_repo
        self._session = db_session  # Direkter Session-Zugriff - Verstoß!

    def get_user_quick(self, user_id: int) -> User:
        # Anfällig: Umgeht Repository
        # Kein Audit-Logging, keine Autorisierungsprüfung
        return self._session.query(User).get(user_id)

    def bulk_update_emails(self, domain_change: dict):
        # Anfällig: Direktes SQL, umgeht alle Sicherheitskontrollen
        self._session.execute(
            "UPDATE users SET email = REPLACE(email, :old, :new)",
            domain_change
        )
        self._session.commit()

Korrigierter Code

// Korrigiert: Aller Datenzugriff nur durch Repository

@Repository
public class UserRepository {
    private final EntityManager em;
    private final AuditService auditService;
    private final CacheService cacheService;

    public User findById(Long id) {
        // Cache zuerst prüfen
        User cached = cacheService.get("user:" + id);
        if (cached != null) return cached;

        auditService.log("user.read", id);
        User user = em.find(User.class, id);

        cacheService.put("user:" + id, user);
        return user;
    }

    public void save(User user) {
        validateUser(user);
        auditService.log("user.save", user.getId());
        em.persist(user);
        cacheService.invalidate("user:" + user.getId());
    }

    public void updateEmail(Long id, String newEmail) {
        // Ordnungsgemäße Aktualisierung durch Repository mit allen Kontrollen
        User user = findById(id);
        if (user != null) {
            user.setEmail(newEmail);
            save(user);
        }
    }
}

// Korrigiert: Service verwendet NUR das Repository
@Service
public class FixedUserService {

    private final UserRepository userRepository;

    // Kein direkter EntityManager-Zugriff!

    public FixedUserService(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    public User getUser(Long id) {
        // Korrigiert: Aller Zugriff durch Repository
        return userRepository.findById(id);
    }

    public void updateUserEmail(Long id, String newEmail) {
        // Korrigiert: Verwendet Repository-Methode
        userRepository.updateEmail(id, newEmail);
    }
}

# Korrigiert: Striktes Datenmanager-Muster

from typing import Optional, List
from abc import ABC, abstractmethod

# Interface für Repository - einziger Weg zum Datenzugriff
class IUserRepository(ABC):
    @abstractmethod
    def get_by_id(self, user_id: int) -> Optional[User]: pass

    @abstractmethod
    def save(self, user: User) -> None: pass

    @abstractmethod
    def find_by_email(self, email: str) -> Optional[User]: pass


# Implementierung mit allen zentralisierten Belangen
class UserRepository(IUserRepository):
    def __init__(self, session_factory, audit_service, cache_service):
        self._session_factory = session_factory
        self._audit = audit_service
        self._cache = cache_service

    def get_by_id(self, user_id: int) -> Optional[User]:
        # Cache prüfen
        cached = self._cache.get(f"user:{user_id}")
        if cached:
            return cached

        # Audit
        self._audit.log("user.read", user_id)

        # Abfrage
        with self._session_factory() as session:
            user = session.query(User).get(user_id)

        # Ergebnis cachen
        if user:
            self._cache.set(f"user:{user_id}", user)

        return user

    def save(self, user: User) -> None:
        self._validate(user)
        self._audit.log("user.save", user.id)

        with self._session_factory() as session:
            session.add(user)
            session.commit()

        self._cache.invalidate(f"user:{user.id}")


# Korrigiert: Service hängt nur vom Repository-Interface ab
class FixedUserService:
    def __init__(self, user_repo: IUserRepository):
        # Nur Repository-Zugriff, keine direkte Session
        self._user_repo = user_repo

    def get_user(self, user_id: int) -> Optional[User]:
        # Aller Zugriff durch Repository
        return self._user_repo.get_by_id(user_id)

    def update_user_email(self, user_id: int, new_email: str):
        user = self._user_repo.get_by_id(user_id)
        if user:
            user.email = new_email
            self._user_repo.save(user)

CVE-Beispiele

Diese CWE ist für direkte CVE-Zuordnung als VERBOTEN markiert, da sie ein architektonisches Qualitätsproblem und keine direkte Sicherheitsschwachstelle darstellt.

Verwandte CWEs

  • CWE-1061: Insufficient Encapsulation (Eltern)
  • CWE-1227: Encapsulation Issues (Kategoriemitglied)
  • CWE-1054: Invocation of a Control Element at an Unnecessarily Deep Horizontal Layer (verwandt)

Referenzen

  1. MITRE Corporation. "CWE-1057: Data Access Operations Outside of Expected Data Manager Component." https://cwe.mitre.org/data/definitions/1057.html

  2. Fowler, Martin. "Patterns of Enterprise Application Architecture" - Repository Pattern.

  3. CISQ. "Automated Source Code Quality Measures."