feat: Enhance CSRF protection and support for CSV file uploads across multiple templates

This commit is contained in:
2026-04-17 18:33:55 +02:00
parent 2f65fba3ae
commit a27639a976
6 changed files with 298 additions and 52 deletions
+205 -39
View File
@@ -37,15 +37,19 @@ from apscheduler.schedulers.background import BackgroundScheduler
from bson.objectid import ObjectId
from urllib.parse import urlparse, urlunparse
import requests
import csv
import ipaddress
import os
import json
import datetime
import time
import traceback
import re
import socket
import io
import html
import logging
import secrets
# QR Code functionality deactivated
# import qrcode
# from qrcode.constants import ERROR_CORRECT_L
@@ -143,6 +147,45 @@ def _set_security_headers(response):
return response
def _get_csrf_token():
token = session.get('_csrf_token')
if not token:
token = secrets.token_urlsafe(32)
session['_csrf_token'] = token
return token
def _is_csrf_exempt_request():
return request.method in {'GET', 'HEAD', 'OPTIONS', 'TRACE'}
def _csrf_error_response(message='CSRF token fehlt oder ist ungültig.'):
if request.is_json or request.path.startswith('/api/') or request.path in {'/download_book_cover', '/proxy_image', '/log_mobile_issue'}:
return jsonify({'error': message}), 400
flash(message, 'error')
return redirect(request.referrer or url_for('home'))
@app.before_request
def _enforce_csrf_protection():
if _is_csrf_exempt_request():
_get_csrf_token()
return None
expected_token = _get_csrf_token()
provided_token = (
request.headers.get('X-CSRFToken')
or request.headers.get('X-CSRF-Token')
or request.form.get('csrf_token')
or request.args.get('csrf_token')
)
if not provided_token or not secrets.compare_digest(provided_token, expected_token):
return _csrf_error_response()
return None
def _get_asset_version():
"""Return a cache-busting asset version tied to deployment state."""
env_version = os.getenv('INVENTAR_ASSET_VERSION', '').strip()
@@ -642,6 +685,7 @@ def inject_version():
"""Inject global template variables."""
is_admin = False
asset_version = _get_asset_version()
csrf_token = _get_csrf_token()
unread_notification_count = 0
if 'username' in session:
try:
@@ -665,6 +709,7 @@ def inject_version():
return {
'APP_VERSION': APP_VERSION,
'ASSET_VERSION': asset_version,
'csrf_token': csrf_token,
'CURRENT_MODULE': current_module,
'school_periods': SCHOOL_PERIODS,
'library_module_enabled': cfg.LIBRARY_MODULE_ENABLED,
@@ -1141,6 +1186,87 @@ def _excel_list(value):
return unique
def _load_tabular_upload(uploaded_file):
"""Load a CSV or XLSX upload and return header row plus data rows."""
filename = (getattr(uploaded_file, 'filename', '') or '').lower()
file_bytes = uploaded_file.read()
uploaded_file.stream.seek(0)
if filename.endswith('.csv'):
for encoding in ('utf-8-sig', 'utf-8', 'latin-1'):
try:
text = file_bytes.decode(encoding)
break
except Exception:
text = None
if text is None:
raise ValueError('CSV-Datei konnte nicht dekodiert werden.')
sample = text[:4096]
try:
dialect = csv.Sniffer().sniff(sample, delimiters=';,\t,|')
except Exception:
dialect = csv.excel
dialect.delimiter = ';' if sample.count(';') >= sample.count(',') else ','
reader = csv.reader(io.StringIO(text), dialect)
rows = [row for row in reader if any(str(cell).strip() for cell in row)]
if not rows:
raise ValueError('CSV-Datei enthält keine Daten.')
return rows[0], rows[1:]
try:
from openpyxl import load_workbook
except Exception as exc:
raise ValueError(f'Excel-Import benötigt openpyxl: {exc}') from exc
workbook = load_workbook(io.BytesIO(file_bytes), data_only=True, read_only=True)
sheet = workbook.active
header_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), None)
if not header_row:
workbook.close()
raise ValueError('Die Datei enthält keine Kopfzeile.')
data_rows = list(sheet.iter_rows(min_row=2, values_only=True))
workbook.close()
return header_row, data_rows
def _is_public_host(hostname):
"""Return True only for hosts that resolve to public IPs."""
if not hostname:
return False
hostname = hostname.strip().lower()
if hostname in {'localhost', '127.0.0.1', '::1'}:
return False
try:
resolved_infos = socket.getaddrinfo(hostname, None)
except Exception:
return False
public_seen = False
for info in resolved_infos:
address = info[4][0]
try:
ip_obj = ipaddress.ip_address(address)
except ValueError:
continue
if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local or ip_obj.is_multicast or ip_obj.is_reserved or ip_obj.is_unspecified:
return False
public_seen = True
return public_seen
def _deny_if_unauthenticated_file_access():
"""Block file-serving routes unless a user is logged in."""
if 'username' not in session:
return Response('Forbidden', status=403)
return None
def _student_card_id_slug(value):
"""Build a compact identifier fragment from a name or class value."""
normalized = _normalize_excel_header(value)
@@ -1190,26 +1316,14 @@ def _upload_student_cards_excel():
return redirect(url_for('student_cards_admin'))
filename_lower = excel_file.filename.lower()
if not filename_lower.endswith('.xlsx'):
flash('Nur .xlsx Dateien werden unterstützt.', 'error')
if not filename_lower.endswith(('.xlsx', '.csv')):
flash('Nur .xlsx oder .csv Dateien werden unterstützt.', 'error')
return redirect(url_for('student_cards_admin'))
try:
from openpyxl import load_workbook
except Exception:
flash('Excel-Import benötigt das Paket openpyxl. Bitte Abhängigkeiten aktualisieren.', 'error')
return redirect(url_for('student_cards_admin'))
try:
workbook = load_workbook(excel_file, data_only=True, read_only=True)
sheet = workbook.active
header_row, data_rows = _load_tabular_upload(excel_file)
except Exception as exc:
flash(f'Excel-Datei konnte nicht gelesen werden: {exc}', 'error')
return redirect(url_for('student_cards_admin'))
header_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), None)
if not header_row:
flash('Die Excel-Datei enthält keine Kopfzeile.', 'error')
flash(f'Datei konnte nicht gelesen werden: {exc}', 'error')
return redirect(url_for('student_cards_admin'))
header_map = {}
@@ -1264,7 +1378,7 @@ def _upload_student_cards_excel():
)
processed_rows = 0
for row_number, row_values in enumerate(sheet.iter_rows(min_row=2, values_only=True), start=2):
for row_number, row_values in enumerate(data_rows, start=2):
processed_rows += 1
if processed_rows > max_rows:
validation_errors.append((row_number, f'Maximal {max_rows} Zeilen pro Datei erlaubt'))
@@ -1396,26 +1510,14 @@ def _upload_excel_items(scope='inventory'):
return redirect(url_for(fallback_route))
filename_lower = excel_file.filename.lower()
if not filename_lower.endswith('.xlsx'):
flash('Nur .xlsx Dateien werden unterstützt.', 'error')
if not filename_lower.endswith(('.xlsx', '.csv')):
flash('Nur .xlsx oder .csv Dateien werden unterstützt.', 'error')
return redirect(url_for(fallback_route))
try:
from openpyxl import load_workbook
except Exception:
flash('Excel-Import benötigt das Paket openpyxl. Bitte Abhängigkeiten aktualisieren.', 'error')
return redirect(url_for(fallback_route))
try:
workbook = load_workbook(excel_file, data_only=True, read_only=True)
sheet = workbook.active
header_row, data_rows = _load_tabular_upload(excel_file)
except Exception as exc:
flash(f'Excel-Datei konnte nicht gelesen werden: {exc}', 'error')
return redirect(url_for(fallback_route))
header_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), None)
if not header_row:
flash('Die Excel-Datei enthält keine Kopfzeile.', 'error')
flash(f'Datei konnte nicht gelesen werden: {exc}', 'error')
return redirect(url_for(fallback_route))
header_map = {}
@@ -1504,7 +1606,7 @@ def _upload_excel_items(scope='inventory'):
planned_item_total = 0
row_limit_exceeded = False
for row_number, row_values in enumerate(sheet.iter_rows(min_row=2, values_only=True), start=2):
for row_number, row_values in enumerate(data_rows, start=2):
processed_rows += 1
if processed_rows > max_rows:
row_limit_exceeded = True
@@ -1711,6 +1813,10 @@ def uploaded_file(filename):
flask.Response: The requested file or placeholder image if not found
"""
try:
denied = _deny_if_unauthenticated_file_access()
if denied:
return denied
# Check production path first (deployed environment)
prod_path = "/opt/Inventarsystem/Web/uploads"
dev_path = app.config['UPLOAD_FOLDER']
@@ -1748,6 +1854,10 @@ def thumbnail_file(filename):
flask.Response: The requested thumbnail file or placeholder image if not found
"""
try:
denied = _deny_if_unauthenticated_file_access()
if denied:
return denied
# Check production path first
prod_path = "/var/Inventarsystem/Web/thumbnails"
dev_path = app.config['THUMBNAIL_FOLDER']
@@ -1783,6 +1893,10 @@ def preview_file(filename):
flask.Response: The requested preview file or placeholder image if not found
"""
try:
denied = _deny_if_unauthenticated_file_access()
if denied:
return denied
# Check production path first
prod_path = "/var/Inventarsystem/Web/previews"
dev_path = app.config['PREVIEW_FOLDER']
@@ -1854,6 +1968,10 @@ def catch_all_files(filename):
flask.Response: The requested file or placeholder image if not found
"""
try:
denied = _deny_if_unauthenticated_file_access()
if denied:
return denied
# Check if the file exists in any of our directories
possible_dirs = [
app.config['UPLOAD_FOLDER'],
@@ -1912,7 +2030,9 @@ def test_connection():
Returns:
dict: Status information including version and status code
"""
return {'status': 'success', 'message': 'Connection successful', 'version': __version__, 'status_code': 200}
if 'username' not in session or not us.check_admin(session['username']):
return {'status': 'forbidden'}, 403
return {'status': 'success', 'message': 'Connection successful', 'status_code': 200}
@app.route('/user_status')
@@ -4865,7 +4985,7 @@ def _soft_delete_item_groups(db, root_item_ids, username):
}
@app.route('/delete_item/<id>', methods=['POST', 'GET'])
@app.route('/delete_item/<id>', methods=['POST'])
def delete_item(id):
"""
Route for deleting inventory items.
@@ -7783,9 +7903,17 @@ def download_book_cover():
if not image_url:
return jsonify({"error": "No image URL provided"}), 400
parsed_url = urlparse(image_url)
if parsed_url.scheme != 'https' or not parsed_url.netloc:
return jsonify({"error": "Only public HTTPS URLs are allowed"}), 400
hostname = parsed_url.hostname or ''
if not _is_public_host(hostname):
return jsonify({"error": "Target host is not allowed"}), 400
# Download the image
response = requests.get(image_url, stream=True, timeout=10)
response = requests.get(image_url, stream=True, timeout=10, allow_redirects=False)
if response.status_code != 200:
return jsonify({"error": f"Failed to download image: Status {response.status_code}"}), 400
@@ -7798,6 +7926,14 @@ def download_book_cover():
return jsonify({
"error": f"Nicht unterstütztes Bildformat: {content_type}. Erlaubte Formate: JPG, JPEG, PNG, GIF"
}), 400
content_length = response.headers.get('Content-Length')
if content_length:
try:
if int(content_length) > 5 * 1024 * 1024:
return jsonify({"error": "Image is too large"}), 413
except ValueError:
pass
# Generate a fully unique filename using UUID
import uuid
@@ -7819,7 +7955,11 @@ def download_book_cover():
filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
with open(filepath, 'wb') as f:
written = 0
for chunk in response.iter_content(chunk_size=8192):
written += len(chunk)
if written > 5 * 1024 * 1024:
return jsonify({"error": "Image is too large"}), 413
f.write(chunk)
return jsonify({
@@ -7845,10 +7985,18 @@ def proxy_image():
url = request.args.get('url')
if not url:
return jsonify({"error": "No URL provided"}), 400
parsed_url = urlparse(url)
if parsed_url.scheme != 'https' or not parsed_url.netloc:
return jsonify({"error": "Only public HTTPS URLs are allowed"}), 400
hostname = parsed_url.hostname or ''
if not _is_public_host(hostname):
return jsonify({"error": "Target host is not allowed"}), 400
try:
# Fetch the image from the external source
response = requests.get(url, stream=True, timeout=5)
response = requests.get(url, stream=True, timeout=5, allow_redirects=False)
# Check if the request was successful
if response.status_code != 200:
@@ -7856,10 +8004,28 @@ def proxy_image():
# Get the content type
content_type = response.headers.get('Content-Type', 'image/jpeg')
if not content_type.lower().startswith('image/'):
return jsonify({"error": "Target URL did not return an image"}), 400
content_length = response.headers.get('Content-Length')
if content_length:
try:
if int(content_length) > 5 * 1024 * 1024:
return jsonify({"error": "Image is too large"}), 413
except ValueError:
pass
payload = bytearray()
for chunk in response.iter_content(chunk_size=8192):
if not chunk:
continue
payload.extend(chunk)
if len(payload) > 5 * 1024 * 1024:
return jsonify({"error": "Image is too large"}), 413
# Return the image data with appropriate headers
return Response(
response=response.content,
response=bytes(payload),
status=200,
headers={
'Content-Type': content_type
+69
View File
@@ -14,6 +14,7 @@
<meta name="mobile-web-app-capable" content="yes">
<meta name="apple-mobile-web-app-capable" content="yes">
<meta name="apple-mobile-web-app-status-bar-style" content="black-translucent">
<meta name="csrf-token" content="{{ csrf_token }}">
<title>{% block title %}Inventarsystem{% endblock %}</title>
{% block head %}
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0-alpha1/dist/css/bootstrap.min.css" rel="stylesheet">
@@ -24,6 +25,74 @@
<link rel="stylesheet" href="{{ url_for('static', filename='css/planned_appointments.css', v=ASSET_VERSION) }}">
<link rel="icon" href="{{ url_for('static', filename='favicon.ico') }}">
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0-alpha1/dist/js/bootstrap.bundle.min.js"></script>
<script>
(function () {
const csrfMeta = document.querySelector('meta[name="csrf-token"]');
const csrfToken = csrfMeta ? csrfMeta.content : '';
if (!csrfToken) {
return;
}
const safeMethods = new Set(['GET', 'HEAD', 'OPTIONS', 'TRACE']);
function sameOrigin(url) {
try {
return new URL(url, window.location.href).origin === window.location.origin;
} catch (error) {
return false;
}
}
function ensureFormToken(form) {
const method = (form.getAttribute('method') || 'GET').toUpperCase();
if (safeMethods.has(method)) {
return;
}
const action = form.getAttribute('action') || window.location.href;
if (!sameOrigin(action)) {
return;
}
let tokenInput = form.querySelector('input[name="csrf_token"]');
if (!tokenInput) {
tokenInput = document.createElement('input');
tokenInput.type = 'hidden';
tokenInput.name = 'csrf_token';
form.appendChild(tokenInput);
}
tokenInput.value = csrfToken;
}
document.addEventListener('submit', function (event) {
const form = event.target;
if (form && form.tagName === 'FORM') {
ensureFormToken(form);
}
}, true);
const originalFetch = window.fetch.bind(window);
window.fetch = function (resource, init) {
const options = init ? { ...init } : {};
const method = (options.method || 'GET').toUpperCase();
const targetUrl = resource instanceof Request ? resource.url : String(resource);
if (!safeMethods.has(method) && sameOrigin(targetUrl)) {
const headers = new Headers(resource instanceof Request ? resource.headers : undefined);
if (options.headers) {
new Headers(options.headers).forEach((value, key) => headers.set(key, value));
}
headers.set('X-CSRFToken', csrfToken);
headers.set('X-Requested-With', 'fetch');
options.headers = headers;
}
return originalFetch(resource, options);
};
document.addEventListener('DOMContentLoaded', function () {
document.querySelectorAll('form[method="post"], form[method="POST"]').forEach(ensureFormToken);
});
})();
</script>
<style>
/* ===== MODULE DETECTION & SETUP ===== */
:root {
+6 -6
View File
@@ -3609,9 +3609,9 @@ document.addEventListener('DOMContentLoaded', ()=>{
:
`<button class="ausleihen disabled-button" disabled>${item.BlockedNow ? 'Reserviert' : 'Ausgeliehen'}</button>`
}
<a href="{{ url_for('delete_item', id='') }}${item._id}" onclick="return confirm('Sind Sie sicher, dass Sie dieses Objekt löschen möchten?')">
<button class="delete-button">Löschen</button>
</a>
<form method="POST" action="{{ url_for('delete_item', id='') }}${item._id}" style="display:inline;" onsubmit="return confirm('Sind Sie sicher, dass Sie dieses Objekt löschen möchten?')">
<button class="delete-button" type="submit">Löschen</button>
</form>
<button class="edit-button" onclick="openEditModalForSelectedUnit('${item._id}', 'specific-item-card-${item._id}')">Bearbeiten</button>
<button class="duplicate-button" onclick="duplicateItem('${item._id}')">Duplizieren</button>
${canScheduleItem ? `<button class="schedule-button" onclick="openScheduleModal('${item._id}')">Termin planen</button>` : ''}
@@ -4420,9 +4420,9 @@ document.addEventListener('DOMContentLoaded', ()=>{
<button class="duplicate-button" onclick="duplicateItem('${item._id}')">Duplizieren</button>
${damageReports.length > 0 ? `<button class="damage-button" onclick="markDamageAsRepaired('${item._id}')">Repariert</button>` : `<button class="damage-button" onclick="registerDamage('${item._id}')">Schaden melden</button>`}
${canScheduleItem ? `<button class="schedule-button" onclick="openScheduleModal('${item._id}')">Termin planen</button>` : ''}
<a href="/delete_item/${item._id}" onclick="return confirm('Sind Sie sicher?')">
<button class="delete-button">Löschen</button>
</a>
<form method="POST" action="/delete_item/${item._id}" style="display:inline;" onsubmit="return confirm('Sind Sie sicher?')">
<button class="delete-button" type="submit">Löschen</button>
</form>
</div>
`;
+2 -2
View File
@@ -239,9 +239,9 @@
<div style="border:1px solid #dbe4ee; border-radius:8px; padding:14px; margin-bottom:16px; background:#f8fbff;">
<h3 style="margin:0 0 8px 0;">Excel-Import Bibliotheksausweise</h3>
<p style="margin:0 0 10px 0; color:#555;">Laden Sie eine <strong>.xlsx</strong>-Datei hoch, zum Beispiel aus <strong>ASV (Amtliche Schuldaten)</strong>. Erkannt werden automatisch Spalten wie <strong>Name</strong>, <strong>Klasse</strong>, <strong>Ausweis-ID</strong>, <strong>Notizen</strong> und <strong>Standard-Ausleihdauer</strong>. Fehlt die Ausweis-ID, wird sie automatisch aus Name und Klasse erzeugt.</p>
<p style="margin:0 0 10px 0; color:#555;">Laden Sie eine <strong>.xlsx</strong>- oder <strong>.csv</strong>-Datei hoch, zum Beispiel aus <strong>ASV (Amtliche Schuldaten)</strong>. Erkannt werden automatisch Spalten wie <strong>Name</strong>, <strong>Klasse</strong>, <strong>Ausweis-ID</strong>, <strong>Notizen</strong> und <strong>Standard-Ausleihdauer</strong>. Fehlt die Ausweis-ID, wird sie automatisch aus Name und Klasse erzeugt.</p>
<form method="POST" action="{{ url_for('upload_student_cards_excel') }}" enctype="multipart/form-data" style="display:flex; gap:10px; flex-wrap:wrap; align-items:center;">
<input type="file" name="student_cards_excel" accept=".xlsx" required>
<input type="file" name="student_cards_excel" accept=".xlsx,.csv" required>
<button type="submit" class="btn btn-secondary" name="excel_action" value="validate">Nur validieren</button>
<button type="submit" class="btn btn-primary" name="excel_action" value="import">Ausweise importieren</button>
</form>
+4 -4
View File
@@ -714,9 +714,9 @@
{% if show_library_features %}
<div style="border:1px solid #dbe4ee; border-radius:8px; padding:14px; margin-bottom:16px; background:#f8fbff;">
<h3 style="margin:0 0 8px 0;">Excel-Import Bibliothek (Mehrere Bücher)</h3>
<p style="margin:0 0 10px 0; color:#555;">Laden Sie eine <strong>.xlsx</strong>-Datei hoch. Spalten werden automatisch erkannt (z.B. Name, Ort, Beschreibung, ISBN, Code, Anzahl). Für den Bibliotheksimport ist eine gültige ISBN je Zeile erforderlich.</p>
<p style="margin:0 0 10px 0; color:#555;">Laden Sie eine <strong>.xlsx</strong>- oder <strong>.csv</strong>-Datei hoch. Spalten werden automatisch erkannt (z.B. Name, Ort, Beschreibung, ISBN, Code, Anzahl). Für den Bibliotheksimport ist eine gültige ISBN je Zeile erforderlich.</p>
<form method="POST" action="{{ url_for('upload_library_excel') }}" enctype="multipart/form-data" style="display:flex; gap:10px; flex-wrap:wrap; align-items:center;">
<input type="file" name="library_excel" accept=".xlsx" required>
<input type="file" name="library_excel" accept=".xlsx,.csv" required>
<button type="submit" class="btn btn-secondary" name="excel_action" value="validate">Nur validieren</button>
<button type="submit" class="btn btn-primary" name="excel_action" value="import">Bibliothek importieren</button>
</form>
@@ -724,9 +724,9 @@
{% else %}
<div style="border:1px solid #dbe4ee; border-radius:8px; padding:14px; margin-bottom:16px; background:#f8fbff;">
<h3 style="margin:0 0 8px 0;">Excel-Import Inventar (Mehrere Artikel)</h3>
<p style="margin:0 0 10px 0; color:#555;">Laden Sie eine <strong>.xlsx</strong>-Datei hoch. Spalten werden automatisch erkannt (z.B. Name, Ort, Beschreibung, Filter1/2/3, Kosten, Jahr, Code, Anzahl).</p>
<p style="margin:0 0 10px 0; color:#555;">Laden Sie eine <strong>.xlsx</strong>- oder <strong>.csv</strong>-Datei hoch. Spalten werden automatisch erkannt (z.B. Name, Ort, Beschreibung, Filter1/2/3, Kosten, Jahr, Code, Anzahl).</p>
<form method="POST" action="{{ url_for('upload_inventory_excel') }}" enctype="multipart/form-data" style="display:flex; gap:10px; flex-wrap:wrap; align-items:center;">
<input type="file" name="inventory_excel" accept=".xlsx" required>
<input type="file" name="inventory_excel" accept=".xlsx,.csv" required>
<button type="submit" class="btn btn-secondary" name="excel_action" value="validate">Nur validieren</button>
<button type="submit" class="btn btn-primary" name="excel_action" value="import">Inventar importieren</button>
</form>
+12 -1
View File
@@ -79,7 +79,18 @@ def check_password_strength(password):
Returns:
bool: True if password is strong enough, False otherwise
"""
if len(password) < 6:
if password is None:
return False
if len(password) < 12:
return False
has_lower = any(char.islower() for char in password)
has_upper = any(char.isupper() for char in password)
has_digit = any(char.isdigit() for char in password)
has_symbol = any(not char.isalnum() for char in password)
if not (has_lower and has_upper and has_digit and has_symbol):
return False
return True