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
| Auswirkung | Details |
|---|---|
| Andere | Bereich: Ändere Reduzierte Wartbarkeit - Verstreuter Datenzugriff macht das System schwerer zu verstehen und zu modifizieren. |
| Integrität | Bereich: Integrität Schutzmechanismus umgehen - Sicherheitskontrollen in der Datenschicht werden umgangen. |
| Andere | Bereich: Ä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
-
MITRE Corporation. "CWE-1083: Data Access from Outside Expected Data Manager Component." https://cwe.mitre.org/data/definitions/1083.html
-
Fowler, Martin. "Patterns of Enterprise Application Architecture." - Repository Pattern
-
Evans, Eric. "Domain-Driven Design." - Layered Architecture