302 lines
10 KiB
Python
302 lines
10 KiB
Python
from flask import Blueprint, request, jsonify, g
|
|
from app import db, limiter, client_ip
|
|
from app.models.vault_item import VaultItem, ItemType
|
|
from app.models.folder import Folder
|
|
from app.models.audit_log import AuditLog
|
|
from app.services.auth_service import require_jwt
|
|
|
|
vault_bp = Blueprint('vault', __name__)
|
|
|
|
VALID_TYPES = {t.value for t in ItemType}
|
|
|
|
|
|
def _validate_folder_id(folder_id, user_id: int):
|
|
"""
|
|
Verify folder_id belongs to user_id.
|
|
Returns the sanitised folder_id (int or None).
|
|
Raises ValueError with a safe message if the folder doesn't exist or
|
|
belongs to another user.
|
|
"""
|
|
if folder_id is None:
|
|
return None
|
|
try:
|
|
folder_id = int(folder_id)
|
|
except (TypeError, ValueError):
|
|
raise ValueError('folder_id must be an integer')
|
|
folder = Folder.query.filter_by(id=folder_id, user_id=user_id).first()
|
|
if not folder:
|
|
raise ValueError('Folder not found')
|
|
return folder_id
|
|
|
|
|
|
@vault_bp.route('', methods=['GET'])
|
|
@limiter.limit('120 per minute')
|
|
@require_jwt
|
|
def list_items():
|
|
items = VaultItem.query.filter_by(user_id=g.current_user_id).order_by(
|
|
VaultItem.created_at.desc()
|
|
).all()
|
|
return jsonify([item.to_dict() for item in items]), 200
|
|
|
|
|
|
@vault_bp.route('', methods=['POST'])
|
|
@limiter.limit('60 per minute')
|
|
@require_jwt
|
|
def create_item():
|
|
data = request.get_json(silent=True) or {}
|
|
name = (data.get('name') or '').strip()
|
|
item_type = data.get('item_type', 'password')
|
|
enc_data = data.get('enc_data', '')
|
|
iv = data.get('iv', '')
|
|
folder_id = data.get('folder_id')
|
|
|
|
enc_name = data.get('enc_name', '')
|
|
iv_name = data.get('iv_name', '')
|
|
|
|
if not name:
|
|
return jsonify({'error': 'name is required'}), 400
|
|
if item_type not in VALID_TYPES:
|
|
return jsonify({'error': f'item_type must be one of {sorted(VALID_TYPES)}'}), 400
|
|
if not enc_data or not iv:
|
|
return jsonify({'error': 'enc_data and iv are required'}), 400
|
|
|
|
try:
|
|
folder_id = _validate_folder_id(folder_id, g.current_user_id)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
|
|
item = VaultItem(
|
|
user_id=g.current_user_id,
|
|
folder_id=folder_id,
|
|
item_type=item_type, # stored as plain string
|
|
name=name,
|
|
enc_data=enc_data,
|
|
iv=iv,
|
|
enc_name=enc_name or None,
|
|
iv_name=iv_name or None,
|
|
)
|
|
try:
|
|
db.session.add(item)
|
|
db.session.flush() # populate item.id before logging
|
|
AuditLog.log(
|
|
user_id=g.current_user_id,
|
|
action='vault_item.create',
|
|
resource_type='vault_item',
|
|
resource_id=item.id,
|
|
detail=f'Created {item_type} item (id={item.id})',
|
|
ip_address=client_ip(),
|
|
)
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return jsonify({'error': f'Database error: {str(e)}'}), 500
|
|
return jsonify(item.to_dict()), 201
|
|
|
|
|
|
@vault_bp.route('/<int:item_id>', methods=['GET'])
|
|
@limiter.limit('120 per minute')
|
|
@require_jwt
|
|
def get_item(item_id):
|
|
item = VaultItem.query.filter_by(id=item_id, user_id=g.current_user_id).first()
|
|
if not item:
|
|
return jsonify({'error': 'Item not found'}), 404
|
|
return jsonify(item.to_dict()), 200
|
|
|
|
|
|
@vault_bp.route('/<int:item_id>', methods=['PUT'])
|
|
@limiter.limit('60 per minute')
|
|
@require_jwt
|
|
def update_item(item_id):
|
|
item = VaultItem.query.filter_by(id=item_id, user_id=g.current_user_id).first()
|
|
if not item:
|
|
return jsonify({'error': 'Item not found'}), 404
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
if 'name' in data:
|
|
name = data['name'].strip()
|
|
if not name:
|
|
return jsonify({'error': 'name cannot be empty'}), 400
|
|
item.name = name
|
|
if 'folder_id' in data:
|
|
try:
|
|
item.folder_id = _validate_folder_id(data['folder_id'], g.current_user_id)
|
|
except ValueError as e:
|
|
return jsonify({'error': str(e)}), 400
|
|
if 'enc_data' in data:
|
|
item.enc_data = data['enc_data']
|
|
if 'iv' in data:
|
|
item.iv = data['iv']
|
|
if 'enc_name' in data:
|
|
item.enc_name = data['enc_name'] or None
|
|
if 'iv_name' in data:
|
|
item.iv_name = data['iv_name'] or None
|
|
|
|
# Explicitly set updated_at — SQLAlchemy's onupdate lambda only fires when
|
|
# it detects a column-level change, which can be skipped if the session
|
|
# dirty-tracking misses a mutation (e.g. only enc_data/iv changed).
|
|
from datetime import datetime, timezone
|
|
item.updated_at = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
try:
|
|
db.session.flush()
|
|
AuditLog.log(
|
|
user_id=g.current_user_id,
|
|
action='vault_item.update',
|
|
resource_type='vault_item',
|
|
resource_id=item.id,
|
|
detail=f'Updated {item.item_type} item (id={item.id})',
|
|
ip_address=client_ip(),
|
|
)
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return jsonify({'error': f'Database error: {str(e)}'}), 500
|
|
|
|
return jsonify(item.to_dict()), 200
|
|
|
|
|
|
@vault_bp.route('/<int:item_id>', methods=['DELETE'])
|
|
@limiter.limit('60 per minute')
|
|
@require_jwt
|
|
def delete_item(item_id):
|
|
item = VaultItem.query.filter_by(id=item_id, user_id=g.current_user_id).first()
|
|
if not item:
|
|
return jsonify({'error': 'Item not found'}), 404
|
|
item_id_saved = item.id
|
|
item_type_saved = item.item_type
|
|
db.session.delete(item)
|
|
db.session.flush()
|
|
AuditLog.log(
|
|
user_id=g.current_user_id,
|
|
action='vault_item.delete',
|
|
resource_type='vault_item',
|
|
resource_id=item_id_saved,
|
|
detail=f'Deleted {item_type_saved} item (id={item_id_saved})',
|
|
ip_address=client_ip(),
|
|
)
|
|
db.session.commit()
|
|
return jsonify({'message': 'Item deleted'}), 200
|
|
|
|
|
|
# ── Import / Export ───────────────────────────────────────────────────────────
|
|
|
|
@vault_bp.route('/export', methods=['GET'])
|
|
@limiter.limit('10 per minute')
|
|
@require_jwt
|
|
def export_items():
|
|
"""
|
|
Return all vault items as an encrypted JSON export payload.
|
|
The client receives raw encrypted blobs and wraps them in a
|
|
signed JSON envelope — the server never sees plaintext.
|
|
Each object: { id, name, item_type, folder_id, enc_data, iv, enc_name, iv_name,
|
|
created_at, updated_at }
|
|
"""
|
|
items = VaultItem.query.filter_by(user_id=g.current_user_id).order_by(
|
|
VaultItem.created_at.desc()
|
|
).all()
|
|
AuditLog.log(
|
|
user_id=g.current_user_id,
|
|
action='vault_item.export',
|
|
resource_type='vault_item',
|
|
resource_id=None,
|
|
detail=f'Exported {len(items)} vault item(s)',
|
|
ip_address=client_ip(),
|
|
)
|
|
db.session.commit()
|
|
return jsonify([item.to_dict() for item in items]), 200
|
|
|
|
|
|
@vault_bp.route('/import', methods=['POST'])
|
|
@limiter.limit('10 per minute')
|
|
@require_jwt
|
|
def import_items():
|
|
"""
|
|
Bulk-import pre-encrypted vault items.
|
|
Accepts a JSON array of objects matching the POST /api/vault schema.
|
|
Items are imported as-is — the server stores encrypted blobs only.
|
|
Duplicate detection is left to the client.
|
|
Returns { imported: N, skipped: N } where skipped = malformed rows.
|
|
"""
|
|
data = request.get_json(silent=True) or []
|
|
if not isinstance(data, list):
|
|
return jsonify({'error': 'Request body must be a JSON array'}), 400
|
|
|
|
imported = 0
|
|
skipped = 0
|
|
for row in data:
|
|
name = (row.get('name') or '').strip()
|
|
item_type = row.get('item_type', 'password')
|
|
enc_data = row.get('enc_data', '')
|
|
iv = row.get('iv', '')
|
|
if not name or item_type not in VALID_TYPES or not enc_data or not iv:
|
|
skipped += 1
|
|
continue
|
|
folder_id = row.get('folder_id')
|
|
try:
|
|
folder_id = _validate_folder_id(folder_id, g.current_user_id)
|
|
except ValueError:
|
|
folder_id = None # invalid/foreign folder — import to root instead of skipping
|
|
enc_name = row.get('enc_name') or None
|
|
iv_name = row.get('iv_name') or None
|
|
item = VaultItem(
|
|
user_id=g.current_user_id,
|
|
folder_id=folder_id,
|
|
item_type=item_type,
|
|
name=name,
|
|
enc_data=enc_data,
|
|
iv=iv,
|
|
enc_name=enc_name,
|
|
iv_name=iv_name,
|
|
)
|
|
db.session.add(item)
|
|
imported += 1
|
|
|
|
if imported:
|
|
db.session.flush()
|
|
AuditLog.log(
|
|
user_id=g.current_user_id,
|
|
action='vault_item.import',
|
|
resource_type='vault_item',
|
|
resource_id=None,
|
|
detail=f'Imported {imported} item(s), skipped {skipped} malformed row(s)',
|
|
ip_address=client_ip(),
|
|
)
|
|
db.session.commit()
|
|
return jsonify({'imported': imported, 'skipped': skipped}), 200
|
|
|
|
@vault_bp.route('/audit-export', methods=['POST'])
|
|
@limiter.limit('30 per minute')
|
|
@require_jwt
|
|
def audit_bulk_export():
|
|
"""
|
|
Record a client-side bulk export in the audit log.
|
|
|
|
The bulk export is built entirely in the browser (no server round-trip),
|
|
so the server calls this endpoint after the download is triggered.
|
|
Accepts a JSON body: { "item_ids": [int, ...] }
|
|
Validates that every supplied ID belongs to the current user before logging.
|
|
"""
|
|
data = request.get_json(silent=True) or {}
|
|
raw_ids = data.get('item_ids', [])
|
|
if not isinstance(raw_ids, list):
|
|
return jsonify({'error': 'item_ids must be an array'}), 400
|
|
|
|
# Validate ownership — only log IDs that belong to the current user.
|
|
valid_ids = [
|
|
item.id for item in VaultItem.query.filter(
|
|
VaultItem.id.in_(raw_ids),
|
|
VaultItem.user_id == g.current_user_id,
|
|
).all()
|
|
]
|
|
|
|
AuditLog.log(
|
|
user_id=g.current_user_id,
|
|
action='vault_item.export_selection',
|
|
resource_type='vault_item',
|
|
resource_id=None,
|
|
detail=f'Bulk exported {len(valid_ids)} selected item(s) (ids: {sorted(valid_ids)[:20]})',
|
|
ip_address=client_ip(),
|
|
)
|
|
db.session.commit()
|
|
return jsonify({'logged': len(valid_ids)}), 200
|