265 lines
8.8 KiB
Python
265 lines
8.8 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 |