Datenzugriff von außerhalb der erwarteten Datenmanagerkomponente

Beschreibung

Datenzugriff von außerhalb der erwarteten Datenmanagerkomponente tritt auf, wenn ein Produkt Datenzugriffsoperationen durchführt, ohne seine vorgesehene Datenmanagerkomponente (wie eine Repository-Schicht, ORM oder Datenzugriffsschicht) zu verwenden. Wenn die Architektur erwartet, dass alle Datenbankoperationen durch eine bestimmte Datenverwaltungskomponente fließen, aber manche Codes diese Struktur umgehen und direkt auf Daten zugreifen, entstehen Architekturverletzungen, die Wartbarkeit, Sicherheitskontrollen und Datenintegrität untergraben.

Risiko

Das Umgehen der vorgesehenen Datenzugriffsschicht hat erhebliche Sicherheitsimplikationen. In der Datenschicht implementierte Sicherheitskontrollen (Audit-Protokollierung, Zugriffskontrolle, Eingabevalidierung) werden umgangen. Sensible Abfragen verwenden möglicherweise keine Parametrisierung, was SQL-Injection ermöglicht. Caching- und Optimierungsstrategien werden umgangen. Datenvalidierung und Geschäftsregeldurchsetzung wird übersprungen. Audit-Trails haben Lücken, wo direkter Zugriff erfolgte. Verbindungsverwaltung und Pooling können umgangen werden, was Ressourcenerschöpfung verursacht. Die inkonsistenten Datenzugriffsmuster erschweren Sicherheitsaudits.

Lösung

Setzen Sie architektonische Grenzen durch Code-Review und statische Analyse durch. Verwenden Sie Dependency Injection, um Datenzugriffskomponenten bereitzustellen. Machen Sie Datenbankverbindungsdetails privat für die Datenschicht. Wenden Sie das Repository-Pattern an, um Datenzugriff zu zentralisieren. Verwenden Sie Linting-Regeln, um direkte Datenbankimporte außerhalb erlaubter Module zu erkennen. Implementieren Sie Architektur-Fitnessfunktionen, um Schichtgrenzen zu verifizieren. Verwenden Sie Zugriffsmodifikatoren und Modulgrenzen, um direkten Datenbankzugriff zu verhindern. Erwägen Sie Datenbankzugriffsrichtlinien, die einschränken, welche Komponenten sich verbinden können. Dokumentieren Sie die erwartete Architektur und schulen Sie Entwickler in den Mustern.

Häufige Auswirkungen

AuswirkungDetails
AndereBereich: Ändere

Reduzierte Wartbarkeit - Verstreuter Datenzugriff macht das System schwerer zu verstehen und zu modifizieren.
IntegritätBereich: Integrität

Schutzmechanismus umgehen - Sicherheitskontrollen in der Datenschicht werden umgangen.
AndereBereich: Ändere

Reduzierte Zuverlässigkeit - Inkonsistenter Datenzugriff kann Datenintegritätsprobleme verursachen.

Beispielcode

Anfälliger Code

// Anfällig: Direkter Datenbankzugriff unter Umgehung der Repository-Schicht

// Erwartete Architektur:
// Controller -> Service -> Repository -> Datenbank
//
// Anfälliger Code greift direkt auf Datenbank zu von Service oder Controller

// Das ordnungsgemäße Repository (sollte verwendet werden)
@Repository
public class UserRepository {
    @PersistenceContext
    private EntityManager em;

    public User findById(Long id) {
        // Ordnungsgemäß: Verwendet parametrisierte Abfrage
        // Ordnungsgemäß: Audit-Protokollierung hier
        // Ordnungsgemäß: Zugriffskontrollprüfungen hier
        auditLog.log("Benutzersuche: " + id);
        return em.find(User.class, id);
    }

    public List<User> search(String criteria) {
        // Ordnungsgemäß: Eingabevalidierung und -bereinigung
        validateSearchCriteria(criteria);
        return em.createQuery("SELECT u FROM User u WHERE u.name LIKE :criteria")
            .setParameter("criteria", "%" + criteria + "%")
            .getResultList();
    }
}

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

    @Autowired
    private UserRepository userRepository;  // Ordnungsgemäße Abhängigkeit

    // Anfällig: Direkter JDBC-Zugriff unter Umgehung des Repository!
    @Autowired
    private DataSource dataSource;

    public User getUser(Long id) {
        // Ordnungsgemäß: Verwendet Repository
        return userRepository.findById(id);
    }

    // Anfällig: Direkter Datenbankzugriff
    public List<User> searchUsers(String criteria) {
        // Umgeht Repository - keine Validierung, keine Audit-Protokollierung!
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement()) {

            // Anfällig: SQL-Injection UND umgeht Sicherheitskontrollen!
            String sql = "SELECT * FROM users WHERE name LIKE '%" + criteria + "%'";
            ResultSet rs = stmt.executeQuery(sql);

            // Keine Audit-Protokollierung
            // Keine Eingabevalidierung
            // Keine Zugriffskontrollprüfungen
            return mapResults(rs);
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
}

// Anfällig: Controller mit direktem Datenbankzugriff
@RestController
public class VulnerableUserController {

    @Autowired
    private JdbcTemplate jdbcTemplate;  // Sollte keinen direkten Zugriff haben!

    @GetMapping("/api/users/{id}")
    public User getUser(@PathVariable Long id) {
        // Anfällig: Umgeht Service- und Repository-Schichten vollständig!
        String sql = "SELECT * FROM users WHERE id = " + id;
        return jdbcTemplate.queryForObject(sql, new UserRowMapper());
    }
}
# Anfällig: Direkter Datenbankzugriff in Python-Anwendung

# Erwartete Architektur verwendet Repository-Pattern
class UserRepository:
    """Ordnungsgemäße Datenzugriffsschicht mit Sicherheitskontrollen."""

    def __init__(self, db_session):
        self._session = db_session
        self._audit_logger = AuditLogger()

    def find_by_id(self, user_id: int) -> Optional[User]:
        self._audit_logger.log(f"Benutzersuche: {user_id}")
        return self._session.query(User).filter(User.id == user_id).first()

    def search(self, criteria: str) -> List[User]:
        # Eingabe validieren
        criteria = self._sanitize(criteria)
        self._audit_logger.log(f"Benutzersuche: {criteria}")
        return self._session.query(User).filter(
            User.name.ilike(f"%{criteria}%")
        ).all()


# Anfällig: Service umgeht Repository
class VulnerableUserService:

    def __init__(self, repository: UserRepository, engine):
        self.repository = repository
        self._engine = engine  # Sollte keinen direkten Zugriff haben!

    def get_user(self, user_id: int) -> User:
        # Ordnungsgemäß: Verwendet Repository
        return self.repository.find_by_id(user_id)

    def search_users(self, criteria: str) -> List[User]:
        # Anfällig: Umgeht Repository!
        import sqlite3

        # Direkte Verbindung - umgeht alle Sicherheitskontrollen
        with self._engine.connect() as conn:
            # SQL-Injection-Schwachstelle!
            result = conn.execute(
                f"SELECT * FROM users WHERE name LIKE '%{criteria}%'"
            )
            return [self._map_row(row) for row in result]


# Anfällig: View/Controller mit direktem Zugriff
from flask import Flask, request
import psycopg2

app = Flask(__name__)

# Anfällig: Globale Datenbankverbindung in View-Schicht!
db_connection = psycopg2.connect("dbname=myapp user=admin password=secret")

@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    # Anfällig: Direkter Datenbankzugriff vom Controller!
    cursor = db_connection.cursor()
    cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")  # SQL-Injection!
    row = cursor.fetchone()

    # Keine Audit-Protokollierung
    # Keine Zugriffskontrolle
    # Kein Connection Pooling
    return jsonify(row)

Korrigierter Code

// Korrigiert: Durchgesetzter Datenzugriff durch Repository-Schicht

// Korrigiert: Repository ist die EINZIGE Komponente mit Datenbankzugriff
@Repository
public class FixedUserRepository {

    @PersistenceContext
    private EntityManager em;

    @Autowired
    private AuditLogger auditLogger;

    @Autowired
    private AccessControlService accessControl;

    public User findById(Long id) {
        // Korrigiert: Alle Sicherheitskontrollen an einem Ort
        accessControl.checkPermission("user:read");
        auditLogger.log("Benutzersuche", Map.of("userId", id));

        return em.find(User.class, id);
    }

    public List<User> search(String criteria, User requestingUser) {
        // Korrigiert: Validieren und bereinigen
        String sanitized = InputValidator.sanitizeSearchTerm(criteria);

        // Korrigiert: Zugriffskontrolle
        accessControl.checkPermission(requestingUser, "user:search");

        // Korrigiert: Audit-Protokollierung
        auditLogger.log("Benutzersuche", Map.of(
            "criteria", sanitized,
            "requestedBy", requestingUser.getId()
        ));

        // Korrigiert: Parametrisierte Abfrage
        return em.createQuery(
            "SELECT u FROM User u WHERE u.name LIKE :criteria", User.class)
            .setParameter("criteria", "%" + sanitized + "%")
            .getResultList();
    }
}

// Korrigiert: Service verwendet nur Repository - kein direkter Datenbankzugriff
@Service
public class FixedUserService {

    private final UserRepository userRepository;

    // Korrigiert: Nur Repository injiziert, keine DataSource
    public FixedUserService(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    public User getUser(Long id) {
        return userRepository.findById(id);
    }

    public List<User> searchUsers(String criteria, User requestingUser) {
        // Korrigiert: Delegiert an Repository - Sicherheitskontrollen angewendet
        return userRepository.search(criteria, requestingUser);
    }
}

// Korrigiert: Controller hat keine Datenbankabhängigkeiten
@RestController
public class FixedUserController {

    private final UserService userService;

    // Korrigiert: Nur Service injiziert
    public FixedUserController(UserService userService) {
        this.userService = userService;
    }

    @GetMapping("/api/users/{id}")
    public ResponseEntity<User> getUser(@PathVariable Long id) {
        User user = userService.getUser(id);
        return user != null
            ? ResponseEntity.ok(user)
            : ResponseEntity.notFound().build();
    }
}

// Korrigiert: Architekdurchsetzung via Modulgrenzen
// In module-info.java oder via package-private Zugriff

// data-access Modul exportiert nur Interfaces
module dataaccess {
    exports com.example.repository;  // Nur Repository-Interfaces
    // Exportiert NICHT: com.example.repository.impl
    // Exportiert NICHT: javax.sql
}

// service Modul hängt nur von Repository-Interfaces ab
module service {
    requires dataaccess;
    // Kann nicht direkt auf Datenbank zugreifen!
}
# Korrigiert: Durchgesetzte Datenzugriffsarchitektur in Python

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


# Korrigiert: Repository-Interface
class UserRepositoryInterface(ABC):
    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional['User']:
        pass

    @abstractmethod
    def search(self, criteria: str, requesting_user: 'User') -> List['User']:
        pass


# Korrigiert: Repository-Implementierung (in separatem Modul)
class UserRepository(UserRepositoryInterface):
    """Repository mit allen Sicherheitskontrollen."""

    def __init__(self, session, audit_logger, access_control):
        self._session = session
        self._audit = audit_logger
        self._access = access_control

    def find_by_id(self, user_id: int) -> Optional[User]:
        self._access.check_permission("user:read")
        self._audit.log("user_lookup", user_id=user_id)

        return self._session.query(User).filter(User.id == user_id).first()

    def search(self, criteria: str, requesting_user: User) -> List[User]:
        # Validieren
        sanitized = InputValidator.sanitize(criteria)

        # Zugriffskontrolle
        self._access.check_permission(requesting_user, "user:search")

        # Audit
        self._audit.log("user_search",
                       criteria=sanitized,
                       requested_by=requesting_user.id)

        # Sichere Abfrage
        return self._session.query(User).filter(
            User.name.ilike(f"%{sanitized}%")
        ).all()


# Korrigiert: Service hängt nur von Repository-Interface ab
class UserService:
    """Service-Schicht - kein Datenbankzugriff."""

    def __init__(self, user_repository: UserRepositoryInterface):
        # Korrigiert: Nur Repository-Interface, keine Datenbankverbindung
        self._repo = user_repository

    def get_user(self, user_id: int) -> Optional[User]:
        return self._repo.find_by_id(user_id)

    def search_users(self, criteria: str, requesting_user: User) -> List[User]:
        return self._repo.search(criteria, requesting_user)


# Korrigiert: Flask-App mit ordnungsgemäßer Dependency Injection
from flask import Flask, g
from dependency_injector import containers, providers


class Container(containers.DeclarativeContainer):
    """Dependency-Injection-Container."""

    config = providers.Configuration()

    # Datenbank-Session - nur für Repository verfügbar
    db_session = providers.Singleton(
        create_session,
        connection_string=config.database_url
    )

    # Repository mit allen Abhängigkeiten
    user_repository = providers.Factory(
        UserRepository,
        session=db_session,
        audit_logger=providers.Factory(AuditLogger),
        access_control=providers.Factory(AccessControlService)
    )

    # Service hängt nur von Repository ab
    user_service = providers.Factory(
        UserService,
        user_repository=user_repository
    )


app = Flask(__name__)
container = Container()


@app.route('/api/users/<int:user_id>')
def get_user(user_id: int):
    # Korrigiert: Verwendet Service, kein direkter Datenbankzugriff
    service = container.user_service()
    user = service.get_user(user_id)

    if user is None:
        return {'error': 'Benutzer nicht gefunden'}, 404

    return user.to_dict()


# Korrigiert: Linting-Regel um direkte Datenbankimporte zu erkennen
# In .flake8 oder setup.cfg:
# [flake8]
# per-file-ignores =
#     # Nur Repository-Modul kann Datenbank importieren
#     app/repository/*.py: DBAccess
#     app/**/*.py: DBAccess:error

CVE-Beispiele

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


Verwandte CWEs

  • CWE-1061: Insufficient Encapsulation (Eltern)
  • CWE-1227: Encapsulation Issues (Kategoriemitglied)
  • CWE-89: SQL Injection (wird häufig durch Umgehen der Datenschicht ermöglicht)

Referenzen

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

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

  3. Evans, Eric. "Domain-Driven Design." - Layered Architecture