515 lines
17 KiB
Python
515 lines
17 KiB
Python
import base64
|
|
|
|
from flask import request, jsonify
|
|
|
|
from app.api import api_bp
|
|
from app.api.auth import token_required
|
|
from app.extensions import db
|
|
from app.models.password_vault import PasswordFolder, PasswordEntry, PasswordShare
|
|
from app.models.user import User
|
|
|
|
|
|
# --- Folders ---
|
|
|
|
@api_bp.route('/passwords/folders', methods=['GET'])
|
|
@token_required
|
|
def list_password_folders():
|
|
user = request.current_user
|
|
own = PasswordFolder.query.filter_by(owner_id=user.id).all()
|
|
|
|
# Get shared folders
|
|
shared_folder_shares = PasswordShare.query.filter_by(
|
|
shared_with_id=user.id, shareable_type='folder'
|
|
).all()
|
|
shared_ids = [s.shareable_id for s in shared_folder_shares]
|
|
shared = PasswordFolder.query.filter(PasswordFolder.id.in_(shared_ids)).all() if shared_ids else []
|
|
|
|
result = []
|
|
for f in own:
|
|
d = f.to_dict()
|
|
d['permission'] = 'owner'
|
|
result.append(d)
|
|
for f in shared:
|
|
d = f.to_dict()
|
|
share = next((s for s in shared_folder_shares if s.shareable_id == f.id), None)
|
|
d['permission'] = share.permission if share else 'read'
|
|
d['owner_name'] = f.owner.username
|
|
result.append(d)
|
|
|
|
return jsonify(result), 200
|
|
|
|
|
|
@api_bp.route('/passwords/folders', methods=['POST'])
|
|
@token_required
|
|
def create_password_folder():
|
|
user = request.current_user
|
|
data = request.get_json()
|
|
name = data.get('name', '').strip()
|
|
if not name:
|
|
return jsonify({'error': 'Name erforderlich'}), 400
|
|
|
|
folder = PasswordFolder(
|
|
owner_id=user.id,
|
|
parent_id=data.get('parent_id'),
|
|
name=name,
|
|
icon=data.get('icon'),
|
|
)
|
|
db.session.add(folder)
|
|
db.session.commit()
|
|
return jsonify(folder.to_dict()), 201
|
|
|
|
|
|
@api_bp.route('/passwords/folders/<int:folder_id>', methods=['PUT'])
|
|
@token_required
|
|
def update_password_folder(folder_id):
|
|
user = request.current_user
|
|
folder = db.session.get(PasswordFolder, folder_id)
|
|
if not folder or folder.owner_id != user.id:
|
|
return jsonify({'error': 'Nicht gefunden'}), 404
|
|
|
|
data = request.get_json()
|
|
if 'name' in data:
|
|
folder.name = data['name'].strip()
|
|
if 'icon' in data:
|
|
folder.icon = data['icon']
|
|
if 'parent_id' in data:
|
|
folder.parent_id = data['parent_id']
|
|
|
|
db.session.commit()
|
|
return jsonify(folder.to_dict()), 200
|
|
|
|
|
|
@api_bp.route('/passwords/folders/<int:folder_id>', methods=['DELETE'])
|
|
@token_required
|
|
def delete_password_folder(folder_id):
|
|
user = request.current_user
|
|
folder = db.session.get(PasswordFolder, folder_id)
|
|
if not folder or folder.owner_id != user.id:
|
|
return jsonify({'error': 'Nicht gefunden'}), 404
|
|
|
|
db.session.delete(folder)
|
|
db.session.commit()
|
|
return jsonify({'message': 'Ordner geloescht'}), 200
|
|
|
|
|
|
# --- Entries ---
|
|
|
|
@api_bp.route('/passwords/entries', methods=['GET'])
|
|
@token_required
|
|
def list_password_entries():
|
|
user = request.current_user
|
|
folder_id = request.args.get('folder_id', None, type=int)
|
|
category = request.args.get('category', None)
|
|
|
|
query = PasswordEntry.query.filter_by(user_id=user.id)
|
|
if folder_id is not None:
|
|
query = query.filter_by(folder_id=folder_id)
|
|
if category:
|
|
query = query.filter_by(category=category)
|
|
|
|
entries = query.order_by(PasswordEntry.created_at.desc()).all()
|
|
|
|
# Also get shared entries
|
|
shared_entry_shares = PasswordShare.query.filter_by(
|
|
shared_with_id=user.id, shareable_type='entry'
|
|
).all()
|
|
shared_ids = [s.shareable_id for s in shared_entry_shares]
|
|
shared = PasswordEntry.query.filter(PasswordEntry.id.in_(shared_ids)).all() if shared_ids else []
|
|
|
|
result = [e.to_dict() for e in entries]
|
|
for e in shared:
|
|
d = e.to_dict()
|
|
d['shared'] = True
|
|
share = next((s for s in shared_entry_shares if s.shareable_id == e.id), None)
|
|
d['permission'] = share.permission if share else 'read'
|
|
result.append(d)
|
|
|
|
return jsonify(result), 200
|
|
|
|
|
|
@api_bp.route('/passwords/entries', methods=['POST'])
|
|
@token_required
|
|
def create_password_entry():
|
|
user = request.current_user
|
|
data = request.get_json()
|
|
|
|
if 'title_encrypted' not in data or 'iv' not in data:
|
|
return jsonify({'error': 'Verschluesselte Daten + IV erforderlich'}), 400
|
|
|
|
entry = PasswordEntry(
|
|
user_id=user.id,
|
|
folder_id=data.get('folder_id'),
|
|
title_encrypted=base64.b64decode(data['title_encrypted']),
|
|
url_encrypted=base64.b64decode(data['url_encrypted']) if data.get('url_encrypted') else None,
|
|
username_encrypted=base64.b64decode(data['username_encrypted']) if data.get('username_encrypted') else None,
|
|
password_encrypted=base64.b64decode(data['password_encrypted']) if data.get('password_encrypted') else None,
|
|
notes_encrypted=base64.b64decode(data['notes_encrypted']) if data.get('notes_encrypted') else None,
|
|
totp_secret_encrypted=base64.b64decode(data['totp_secret_encrypted']) if data.get('totp_secret_encrypted') else None,
|
|
passkey_data_encrypted=base64.b64decode(data['passkey_data_encrypted']) if data.get('passkey_data_encrypted') else None,
|
|
iv=base64.b64decode(data['iv']),
|
|
category=data.get('category'),
|
|
)
|
|
db.session.add(entry)
|
|
db.session.commit()
|
|
return jsonify(entry.to_dict()), 201
|
|
|
|
|
|
@api_bp.route('/passwords/entries/<int:entry_id>', methods=['PUT'])
|
|
@token_required
|
|
def update_password_entry(entry_id):
|
|
user = request.current_user
|
|
entry = db.session.get(PasswordEntry, entry_id)
|
|
if not entry:
|
|
return jsonify({'error': 'Nicht gefunden'}), 404
|
|
|
|
# Check access
|
|
if entry.user_id != user.id:
|
|
share = PasswordShare.query.filter_by(
|
|
shareable_type='entry', shareable_id=entry_id, shared_with_id=user.id
|
|
).first()
|
|
if not share or share.permission not in ('write', 'manage'):
|
|
return jsonify({'error': 'Zugriff verweigert'}), 403
|
|
|
|
data = request.get_json()
|
|
for field in ['title_encrypted', 'url_encrypted', 'username_encrypted',
|
|
'password_encrypted', 'notes_encrypted', 'totp_secret_encrypted',
|
|
'passkey_data_encrypted', 'iv']:
|
|
if field in data and data[field]:
|
|
setattr(entry, field, base64.b64decode(data[field]))
|
|
|
|
if 'category' in data:
|
|
entry.category = data['category']
|
|
if 'folder_id' in data:
|
|
entry.folder_id = data['folder_id']
|
|
|
|
db.session.commit()
|
|
return jsonify(entry.to_dict()), 200
|
|
|
|
|
|
@api_bp.route('/passwords/entries/<int:entry_id>', methods=['DELETE'])
|
|
@token_required
|
|
def delete_password_entry(entry_id):
|
|
user = request.current_user
|
|
entry = db.session.get(PasswordEntry, entry_id)
|
|
if not entry:
|
|
return jsonify({'error': 'Nicht gefunden'}), 404
|
|
if entry.user_id != user.id:
|
|
return jsonify({'error': 'Zugriff verweigert'}), 403
|
|
|
|
db.session.delete(entry)
|
|
db.session.commit()
|
|
return jsonify({'message': 'Eintrag geloescht'}), 200
|
|
|
|
|
|
# --- Sharing ---
|
|
|
|
@api_bp.route('/passwords/share', methods=['POST'])
|
|
@token_required
|
|
def share_password():
|
|
user = request.current_user
|
|
data = request.get_json()
|
|
|
|
shareable_type = data.get('type') # 'entry' or 'folder'
|
|
shareable_id = data.get('id')
|
|
username = data.get('username', '').strip()
|
|
permission = data.get('permission', 'read')
|
|
|
|
if shareable_type not in ('entry', 'folder'):
|
|
return jsonify({'error': 'Typ muss "entry" oder "folder" sein'}), 400
|
|
|
|
if permission not in ('read', 'write', 'manage'):
|
|
return jsonify({'error': 'Ungueltige Berechtigung'}), 400
|
|
|
|
# Verify ownership
|
|
if shareable_type == 'entry':
|
|
obj = db.session.get(PasswordEntry, shareable_id)
|
|
if not obj or obj.user_id != user.id:
|
|
return jsonify({'error': 'Nicht gefunden'}), 404
|
|
else:
|
|
obj = db.session.get(PasswordFolder, shareable_id)
|
|
if not obj or obj.owner_id != user.id:
|
|
return jsonify({'error': 'Nicht gefunden'}), 404
|
|
|
|
target = User.query.filter_by(username=username).first()
|
|
if not target:
|
|
return jsonify({'error': 'Benutzer nicht gefunden'}), 404
|
|
if target.id == user.id:
|
|
return jsonify({'error': 'Kann nicht mit sich selbst teilen'}), 400
|
|
|
|
existing = PasswordShare.query.filter_by(
|
|
shareable_type=shareable_type, shareable_id=shareable_id,
|
|
shared_with_id=target.id
|
|
).first()
|
|
if existing:
|
|
existing.permission = permission
|
|
else:
|
|
share = PasswordShare(
|
|
shareable_type=shareable_type,
|
|
shareable_id=shareable_id,
|
|
shared_by_id=user.id,
|
|
shared_with_id=target.id,
|
|
permission=permission,
|
|
encrypted_key=base64.b64decode(data['encrypted_key']) if data.get('encrypted_key') else None,
|
|
)
|
|
db.session.add(share)
|
|
|
|
db.session.commit()
|
|
return jsonify({'message': f'Mit {username} geteilt'}), 200
|
|
|
|
|
|
@api_bp.route('/passwords/shares', methods=['GET'])
|
|
@token_required
|
|
def list_password_shares():
|
|
user = request.current_user
|
|
shareable_type = request.args.get('type')
|
|
shareable_id = request.args.get('id', type=int)
|
|
|
|
query = PasswordShare.query.filter_by(shared_by_id=user.id)
|
|
if shareable_type:
|
|
query = query.filter_by(shareable_type=shareable_type)
|
|
if shareable_id:
|
|
query = query.filter_by(shareable_id=shareable_id)
|
|
|
|
shares = query.all()
|
|
return jsonify([{
|
|
'id': s.id,
|
|
'type': s.shareable_type,
|
|
'shareable_id': s.shareable_id,
|
|
'shared_with': s.shared_with.username,
|
|
'permission': s.permission,
|
|
} for s in shares]), 200
|
|
|
|
|
|
@api_bp.route('/passwords/shares/<int:share_id>', methods=['DELETE'])
|
|
@token_required
|
|
def remove_password_share(share_id):
|
|
user = request.current_user
|
|
share = db.session.get(PasswordShare, share_id)
|
|
if not share or share.shared_by_id != user.id:
|
|
return jsonify({'error': 'Nicht gefunden'}), 404
|
|
|
|
db.session.delete(share)
|
|
db.session.commit()
|
|
return jsonify({'message': 'Freigabe entfernt'}), 200
|
|
|
|
|
|
# --- KeePass Import ---
|
|
|
|
@api_bp.route('/passwords/import/keepass', methods=['POST'])
|
|
@token_required
|
|
def import_keepass():
|
|
user = request.current_user
|
|
|
|
if 'file' not in request.files:
|
|
return jsonify({'error': 'Keine Datei gesendet'}), 400
|
|
|
|
kdbx_file = request.files['file']
|
|
kdbx_password = request.form.get('password', '')
|
|
|
|
if not kdbx_password:
|
|
return jsonify({'error': 'KeePass-Passwort erforderlich'}), 400
|
|
|
|
try:
|
|
from pykeepass import PyKeePass
|
|
import tempfile
|
|
import os
|
|
|
|
# Save to temp file
|
|
with tempfile.NamedTemporaryFile(delete=False, suffix='.kdbx') as tmp:
|
|
kdbx_file.save(tmp.name)
|
|
tmp_path = tmp.name
|
|
|
|
try:
|
|
kp = PyKeePass(tmp_path, password=kdbx_password)
|
|
finally:
|
|
os.unlink(tmp_path)
|
|
|
|
# Return entries as plaintext - frontend will encrypt them
|
|
entries = []
|
|
groups = []
|
|
|
|
for group in kp.groups:
|
|
if group.name and group.name not in ('Root', 'Recycle Bin'):
|
|
groups.append({
|
|
'name': group.name,
|
|
'path': '/'.join(g.name for g in group.path if g.name),
|
|
'uuid': str(group.uuid),
|
|
'parent_uuid': str(group.parentgroup.uuid) if group.parentgroup else None,
|
|
})
|
|
|
|
for entry in kp.entries:
|
|
if entry.title:
|
|
group_path = '/'.join(g.name for g in entry.group.path if g.name) if entry.group else ''
|
|
entries.append({
|
|
'title': entry.title or '',
|
|
'url': entry.url or '',
|
|
'username': entry.username or '',
|
|
'password': entry.password or '',
|
|
'notes': entry.notes or '',
|
|
'totp': entry.otp or '',
|
|
'group': group_path,
|
|
'group_uuid': str(entry.group.uuid) if entry.group else None,
|
|
})
|
|
|
|
return jsonify({
|
|
'entries': entries,
|
|
'groups': groups,
|
|
'count': len(entries),
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({'error': f'Import fehlgeschlagen: {str(e)}'}), 400
|
|
|
|
|
|
# --- Firefox CSV Import ---
|
|
|
|
@api_bp.route('/passwords/import/firefox', methods=['POST'])
|
|
@token_required
|
|
def import_firefox():
|
|
"""Import passwords from Firefox CSV export.
|
|
|
|
Firefox: Einstellungen > Passwoerter > ... > Passwoerter exportieren
|
|
CSV columns: url, username, password, httpRealm, formActionOrigin, guid, timeCreated, timeLastUsed, timePasswordChanged
|
|
"""
|
|
user = request.current_user
|
|
|
|
if 'file' not in request.files:
|
|
return jsonify({'error': 'Keine Datei gesendet'}), 400
|
|
|
|
csv_file = request.files['file']
|
|
|
|
try:
|
|
import csv
|
|
import io
|
|
|
|
content = csv_file.read().decode('utf-8')
|
|
reader = csv.DictReader(io.StringIO(content))
|
|
|
|
entries = []
|
|
for row in reader:
|
|
url = row.get('url', row.get('origin', '')).strip()
|
|
username = row.get('username', '').strip()
|
|
password = row.get('password', '').strip()
|
|
|
|
if not url and not username and not password:
|
|
continue
|
|
|
|
# Extract domain as title
|
|
title = url
|
|
try:
|
|
from urllib.parse import urlparse
|
|
parsed = urlparse(url)
|
|
title = parsed.netloc or parsed.path or url
|
|
except Exception:
|
|
pass
|
|
|
|
entries.append({
|
|
'title': title,
|
|
'url': url,
|
|
'username': username,
|
|
'password': password,
|
|
'notes': '',
|
|
'totp': '',
|
|
'group': 'Firefox Import',
|
|
})
|
|
|
|
return jsonify({
|
|
'entries': entries,
|
|
'groups': [{'name': 'Firefox Import', 'uuid': 'firefox-import', 'parent_uuid': None}] if entries else [],
|
|
'count': len(entries),
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({'error': f'CSV-Import fehlgeschlagen: {str(e)}'}), 400
|
|
|
|
|
|
# --- Generic CSV Import (Chrome, Bitwarden, etc.) ---
|
|
|
|
@api_bp.route('/passwords/import/csv', methods=['POST'])
|
|
@token_required
|
|
def import_generic_csv():
|
|
"""Import passwords from generic CSV (Chrome, Bitwarden, 1Password, etc.)
|
|
|
|
Tries to auto-detect columns: name/title, url, username, password, notes
|
|
"""
|
|
user = request.current_user
|
|
|
|
if 'file' not in request.files:
|
|
return jsonify({'error': 'Keine Datei gesendet'}), 400
|
|
|
|
csv_file = request.files['file']
|
|
|
|
try:
|
|
import csv
|
|
import io
|
|
|
|
content = csv_file.read().decode('utf-8')
|
|
reader = csv.DictReader(io.StringIO(content))
|
|
|
|
# Map common column names
|
|
col_map = {
|
|
'title': ['title', 'name', 'titel', 'bezeichnung', 'entry'],
|
|
'url': ['url', 'uri', 'website', 'login_uri', 'origin'],
|
|
'username': ['username', 'user', 'login', 'benutzername', 'email', 'login_username'],
|
|
'password': ['password', 'passwort', 'pass', 'login_password'],
|
|
'notes': ['notes', 'note', 'notizen', 'comment', 'comments', 'extra'],
|
|
'totp': ['totp', 'otp', 'login_totp', '2fa'],
|
|
'group': ['group', 'folder', 'ordner', 'category', 'kategorie', 'type'],
|
|
}
|
|
|
|
def find_col(fieldnames, target):
|
|
for col_name in col_map.get(target, []):
|
|
for fn in fieldnames:
|
|
if fn.lower().strip() == col_name:
|
|
return fn
|
|
return None
|
|
|
|
fieldnames = reader.fieldnames or []
|
|
mapping = {target: find_col(fieldnames, target) for target in col_map}
|
|
|
|
entries = []
|
|
groups_set = set()
|
|
for row in reader:
|
|
def get(target):
|
|
col = mapping.get(target)
|
|
return row.get(col, '').strip() if col else ''
|
|
|
|
title = get('title')
|
|
url = get('url')
|
|
username = get('username')
|
|
password = get('password')
|
|
|
|
if not title and not url and not username and not password:
|
|
continue
|
|
|
|
if not title and url:
|
|
try:
|
|
from urllib.parse import urlparse
|
|
title = urlparse(url).netloc or url
|
|
except Exception:
|
|
title = url
|
|
|
|
group = get('group') or 'CSV Import'
|
|
groups_set.add(group)
|
|
|
|
entries.append({
|
|
'title': title or '(Unbenannt)',
|
|
'url': url,
|
|
'username': username,
|
|
'password': password,
|
|
'notes': get('notes'),
|
|
'totp': get('totp'),
|
|
'group': group,
|
|
})
|
|
|
|
groups = [{'name': g, 'uuid': g, 'parent_uuid': None} for g in groups_set]
|
|
|
|
return jsonify({
|
|
'entries': entries,
|
|
'groups': groups,
|
|
'count': len(entries),
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({'error': f'CSV-Import fehlgeschlagen: {str(e)}'}), 400
|