nngo/PassKeeper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

05/02/2026 updated code for security

Browse Source
This commit is contained in:
nngo
2026-05-02 15:30:33 -04:00
parent 9af7435cab
commit c7b1806ec8
6 changed files with 594 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routes/auth.py
+227 -53
View File
@@ -1,7 +1,7 @@
import re
import time
from flask import Blueprint, request, jsonify, g
from flask import Blueprint, request, jsonify, g, session
from app import db, limiter
from app.models.user import User
from app.models.audit_log import AuditLog
@@ -15,6 +15,10 @@ from app.services.auth_service import (
require_jwt,
encrypt_totp_secret,
decrypt_totp_secret,
generate_recovery_nonce,
verify_recovery_proof,
generate_backup_codes,
verify_and_consume_backup_code,
)
auth_bp = Blueprint('auth', __name__)
@@ -66,6 +70,12 @@ def register():
@auth_bp.route('/login', methods=['POST'])
@limiter.limit('10 per minute')
def login():
from datetime import datetime, timezone
# Number of consecutive failures before a temporary lockout is applied.
MAX_FAILED_LOGINS = 5
LOCKOUT_MINUTES = 15
data = request.get_json(silent=True) or {}
email = (data.get('email') or '').strip().lower()
auth_hash = data.get('auth_hash', '')
@@ -76,20 +86,59 @@ def login():
return jsonify({'error': 'Email and auth_hash are required'}), 400
user = User.query.filter_by(email=email).first()
if not user or not verify_auth_token(auth_hash, user.master_hash):
if user:
# Per-account lockout check — evaluated before password verification so the
# check itself doesn't leak whether the account exists via timing.
if user and user.locked_until:
now = datetime.now(timezone.utc).replace(tzinfo=None)
if user.locked_until > now:
remaining = int((user.locked_until - now).total_seconds() // 60) + 1
AuditLog.log(
user_id=user.id,
action='auth.login_failed',
action='auth.login_blocked',
resource_type='user',
resource_id=user.id,
detail='Failed login attempt — invalid password',
detail=f'Login blocked — account locked for {remaining} more minute(s)',
ip_address=_client_ip(),
)
db.session.commit()
return jsonify({
'error': f'Account temporarily locked. Try again in {remaining} minute(s).'
}), 429
else:
# Lockout has expired — reset the counter.
user.failed_login_count = 0
user.locked_until = None
if not user or not verify_auth_token(auth_hash, user.master_hash):
if user:
user.failed_login_count = (user.failed_login_count or 0) + 1
if user.failed_login_count >= MAX_FAILED_LOGINS:
from datetime import timedelta
user.locked_until = datetime.utcnow() + timedelta(minutes=LOCKOUT_MINUTES)
AuditLog.log(
user_id=user.id,
action='auth.account_locked',
resource_type='user',
resource_id=user.id,
detail=f'Account locked for {LOCKOUT_MINUTES} minutes after {user.failed_login_count} failed attempts',
ip_address=_client_ip(),
)
else:
AuditLog.log(
user_id=user.id,
action='auth.login_failed',
resource_type='user',
resource_id=user.id,
detail=f'Failed login attempt — invalid password ({user.failed_login_count}/{MAX_FAILED_LOGINS})',
ip_address=_client_ip(),
)
db.session.commit()
return jsonify({'error': 'Invalid email or password'}), 401
from datetime import datetime
# Successful authentication — reset lockout state.
user.failed_login_count = 0
user.locked_until = None
user.last_login = datetime.utcnow()
AuditLog.log(
@@ -192,7 +241,8 @@ def mfa_setup():
@auth_bp.route('/mfa/enable', methods=['POST'])
@require_jwt
def mfa_enable():
"""Enable MFA after verifying the first TOTP code."""
"""Enable MFA after verifying the first TOTP code. Returns one-time backup codes."""
import json
user = db.session.get(User, g.current_user_id)
if user.totp_enabled:
return jsonify({'error': 'MFA is already enabled'}), 400
@@ -213,45 +263,66 @@ def mfa_enable():
user.totp_iv = totp_iv
user.totp_enabled = True
# Generate one-time backup codes — plaintext shown once, only hashes stored.
plaintext_codes, hashed_codes = generate_backup_codes()
user.mfa_backup_codes = json.dumps(hashed_codes)
AuditLog.log(
user_id=user.id,
action='auth.mfa_enable',
resource_type='user',
resource_id=user.id,
detail='TOTP two-factor authentication enabled',
detail='TOTP two-factor authentication enabled; backup codes generated',
ip_address=_client_ip(),
)
db.session.commit()
return jsonify({'message': 'MFA enabled successfully'}), 200
return jsonify({
'message': 'MFA enabled successfully',
'backup_codes': plaintext_codes,
}), 200
@auth_bp.route('/mfa/disable', methods=['POST'])
@require_jwt
def mfa_disable():
"""Disable MFA after verifying the current TOTP code."""
"""Disable MFA after verifying the current TOTP code or a backup code."""
import json
user = db.session.get(User, g.current_user_id)
if not user.totp_enabled:
return jsonify({'error': 'MFA is not enabled'}), 400
data = request.get_json(silent=True) or {}
totp_code = (data.get('totp_code') or '').strip()
backup_code = (data.get('backup_code') or '').strip().lower().replace('-', '').replace(' ', '')
import pyotp
plaintext_secret = decrypt_totp_secret(user.totp_secret, user.totp_iv)
if not pyotp.TOTP(plaintext_secret).verify(totp_code, valid_window=1):
verified = False
if totp_code:
verified = pyotp.TOTP(plaintext_secret).verify(totp_code, valid_window=1)
elif backup_code:
stored = json.loads(user.mfa_backup_codes or '[]')
matched, remaining = verify_and_consume_backup_code(stored, backup_code)
if matched:
user.mfa_backup_codes = json.dumps(remaining)
verified = True
if not verified:
return jsonify({'error': 'Invalid verification code'}), 400
user.totp_secret = None
user.totp_iv = None
user.totp_enabled = False
user.mfa_backup_codes = None
AuditLog.log(
user_id=user.id,
action='auth.mfa_disable',
resource_type='user',
resource_id=user.id,
detail='TOTP two-factor authentication disabled',
detail='TOTP two-factor authentication disabled; backup codes cleared',
ip_address=_client_ip(),
)
db.session.commit()
@@ -262,13 +333,15 @@ def mfa_disable():
@auth_bp.route('/mfa/verify', methods=['POST'])
@limiter.limit('10 per minute')
def mfa_verify():
"""Complete MFA login: verify TOTP code and exchange mfa_token for real tokens."""
"""Complete MFA login: verify TOTP code (or backup code) and exchange mfa_token for real tokens."""
import json
data = request.get_json(silent=True) or {}
mfa_token = data.get('mfa_token', '')
totp_code = (data.get('totp_code') or '').strip()
backup_code = (data.get('backup_code') or '').strip().lower().replace('-', '').replace(' ', '')
if not mfa_token or not totp_code:
return jsonify({'error': 'mfa_token and totp_code are required'}), 400
if not mfa_token or (not totp_code and not backup_code):
return jsonify({'error': 'mfa_token and either totp_code or backup_code are required'}), 400
try:
payload = decode_token(mfa_token, expected_type='mfa', check_blacklist=True)
@@ -281,7 +354,27 @@ def mfa_verify():
import pyotp
plaintext_secret = decrypt_totp_secret(user.totp_secret, user.totp_iv)
if not pyotp.TOTP(plaintext_secret).verify(totp_code, valid_window=1):
verified = False
if totp_code:
verified = pyotp.TOTP(plaintext_secret).verify(totp_code, valid_window=1)
if not verified and backup_code:
stored = json.loads(user.mfa_backup_codes or '[]')
matched, remaining = verify_and_consume_backup_code(stored, backup_code)
if matched:
user.mfa_backup_codes = json.dumps(remaining)
verified = True
AuditLog.log(
user_id=user.id,
action='auth.mfa_backup_code_used',
resource_type='user',
resource_id=user.id,
detail=f'MFA backup code used; {len(remaining)} code(s) remaining',
ip_address=_client_ip(),
)
if not verified:
return jsonify({'error': 'Invalid verification code'}), 400
# One-time use: blacklist the mfa_token
@@ -308,7 +401,54 @@ def mfa_verify():
@require_jwt
def mfa_status():
user = db.session.get(User, g.current_user_id)
return jsonify({'totp_enabled': user.totp_enabled}), 200
import json
stored = json.loads(user.mfa_backup_codes or '[]')
return jsonify({
'totp_enabled': user.totp_enabled,
'backup_codes_remaining': len(stored),
}), 200
@auth_bp.route('/mfa/backup-codes/regenerate', methods=['POST'])
@require_jwt
def mfa_backup_codes_regenerate():
"""
Regenerate MFA backup codes. Requires a valid TOTP code to authorise.
All existing backup codes are invalidated and replaced.
Returns the new plaintext codes — shown once, never stored.
"""
import json
user = db.session.get(User, g.current_user_id)
if not user.totp_enabled:
return jsonify({'error': 'MFA is not enabled'}), 400
data = request.get_json(silent=True) or {}
totp_code = (data.get('totp_code') or '').strip()
if not totp_code:
return jsonify({'error': 'totp_code is required'}), 400
import pyotp
plaintext_secret = decrypt_totp_secret(user.totp_secret, user.totp_iv)
if not pyotp.TOTP(plaintext_secret).verify(totp_code, valid_window=1):
return jsonify({'error': 'Invalid verification code'}), 400
plaintext_codes, hashed_codes = generate_backup_codes()
user.mfa_backup_codes = json.dumps(hashed_codes)
AuditLog.log(
user_id=user.id,
action='auth.mfa_backup_codes_regenerated',
resource_type='user',
resource_id=user.id,
detail='MFA backup codes regenerated — previous codes invalidated',
ip_address=_client_ip(),
)
db.session.commit()
return jsonify({
'message': 'Backup codes regenerated. Save these — they will not be shown again.',
'backup_codes': plaintext_codes,
}), 200
@auth_bp.route('/me', methods=['GET'])
@@ -521,52 +661,56 @@ def recover_account():
Recover account access using a recovery code.
Flow:
1. Client looks up enc_key_salt and recovery blobs by email.
2. Client decrypts enc_key_salt using the recovery key (derived from the recovery code).
3. Client derives new auth_hash and new vault key with a new master password.
4. Client re-encrypts all vault items with the new vault key.
1. Client calls /recovery/data → receives enc_key_salt, recovery blobs, nonce.
2. Client decrypts recovery_enc_salt using the recovery key → gets enc_key_salt.
3. Client computes: recovery_proof = HMAC-SHA256(enc_key_salt_bytes, nonce).
4. Client derives new credentials and re-encrypts all vault items.
5. Client POSTs everything here in one atomic payload.
This endpoint is unauthenticated — the recovery code is the credential.
The server validates recovery_proof against the value precomputed during
/recovery/data — enc_key_salt is never sent in plaintext.
The nonce is consumed on first use to prevent replay.
"""
data = request.get_json(silent=True) or {}
email = (data.get('email') or '').strip().lower()
new_auth_hash = data.get('new_auth_hash', '')
new_enc_key_salt = data.get('new_enc_key_salt', '')
# Proof that the client successfully decrypted enc_key_salt:
# the client re-derives auth_hash from the *original* enc_key_salt path
# and sends it alongside the new credentials for server-side verification.
recovery_proof = data.get('recovery_proof', '')
client_proof = data.get('recovery_proof', '')
items = data.get('items', [])
if not all([email, new_auth_hash, new_enc_key_salt, recovery_proof]):
if not all([email, new_auth_hash, new_enc_key_salt, client_proof]):
return jsonify({'error': 'email, new_auth_hash, new_enc_key_salt, and recovery_proof are required'}), 400
time.sleep(0.1) # timing mitigation
# Validate session binding.
expected_proof = session.get('recovery_expected_proof', '')
session_user_id = session.get('recovery_user_id')
if not expected_proof or not session_user_id:
return jsonify({'error': 'No active recovery challenge. Call /recovery/data first.'}), 400
user = User.query.filter_by(email=email).first()
if not user or not user.recovery_enc_salt:
if not user or not user.recovery_enc_salt or user.id != session_user_id:
return jsonify({'error': 'No recovery code found for this account'}), 404
# recovery_proof is the enc_key_salt re-encrypted by the client using the
# recovery key — we return it as a blob for the client to verify, then
# the client sends back the decrypted enc_key_salt as recovery_proof.
# Simpler: recovery_proof = HMAC or simply the decrypted enc_key_salt itself,
# which the client proves by sending it back plaintext. The server checks it
# matches user.enc_key_salt — if the recovery code was wrong, decryption
# would produce garbage that won't match.
if recovery_proof != user.enc_key_salt:
if not verify_recovery_proof(expected_proof, client_proof):
AuditLog.log(
user_id=user.id,
action='auth.recovery_failed',
resource_type='user',
resource_id=user.id,
detail='Recovery attempt failed — incorrect recovery code',
detail='Recovery attempt failed — incorrect recovery proof',
ip_address=_client_ip(),
)
db.session.commit()
return jsonify({'error': 'Invalid recovery code'}), 401
# Consume the nonce — one-time use only.
session.pop('recovery_nonce', None)
session.pop('recovery_expected_proof', None)
session.pop('recovery_user_id', None)
try:
from app.models.vault_item import VaultItem
@@ -589,7 +733,6 @@ def recover_account():
if vault_item:
vault_item.enc_data = enc_data
vault_item.iv = iv
# Re-encrypt the name ciphertext if the client sent updated enc_name/iv_name.
if item_data.get('enc_name'):
vault_item.enc_name = item_data['enc_name']
if item_data.get('iv_name'):
@@ -597,7 +740,7 @@ def recover_account():
user.master_hash = hash_auth_token(new_auth_hash)
user.enc_key_salt = new_enc_key_salt
# Recovery code is consumed — clear it so it cannot be reused
# Recovery code is consumed — clear it so it cannot be reused.
user.recovery_enc_salt = None
user.recovery_iv = None
@@ -628,9 +771,16 @@ def recover_account():
def recovery_data():
"""
Return the data the client needs to attempt recovery (unauthenticated).
Exposes only: enc_key_salt, recovery_enc_salt, recovery_iv.
Returns 404 if no recovery code is configured (prevents user enumeration
of which accounts have recovery set up — same response for unknown email).
Exposes: enc_key_salt, recovery_enc_salt, recovery_iv, and a one-time nonce.
The nonce is used for the HMAC-SHA256 challenge-response proof:
- Client decrypts recovery_enc_salt → gets enc_key_salt bytes.
- Client computes: proof = HMAC-SHA256(key=enc_key_salt_bytes, msg=nonce)
- Server stores expected proof in server-side session at challenge time,
verifying it on /recover and /recovery/items without ever receiving
enc_key_salt in plaintext.
Returns 404 if no recovery code is configured (prevents user enumeration).
"""
email = (request.args.get('email') or '').strip().lower()
if not email:
@@ -640,10 +790,28 @@ def recovery_data():
if not user or not user.recovery_enc_salt:
return jsonify({'error': 'No recovery data found'}), 404
# Generate a fresh nonce and precompute the expected HMAC using the stored
# enc_key_salt. The client must return HMAC-SHA256(enc_key_salt, nonce).
# This proves it decrypted the recovery blob correctly without sending
# enc_key_salt in plaintext.
import hashlib, hmac as _hmac
nonce = generate_recovery_nonce()
expected_proof = _hmac.new(
user.enc_key_salt.encode(),
nonce.encode(),
hashlib.sha256,
).hexdigest()
# Store expected proof and bind it to the user — consumed on first use.
session['recovery_nonce'] = nonce
session['recovery_expected_proof'] = expected_proof
session['recovery_user_id'] = user.id
return jsonify({
'enc_key_salt': user.enc_key_salt,
'recovery_enc_salt': user.recovery_enc_salt,
'recovery_iv': user.recovery_iv,
'nonce': nonce,
}), 200
@@ -653,24 +821,31 @@ def recovery_items():
"""
Return encrypted vault items for recovery re-encryption (unauthenticated).
Requires X-Recovery-Proof header containing the plaintext enc_key_salt.
The server verifies it matches user.enc_key_salt — proof that the client
correctly decrypted the recovery blob (i.e. has the correct recovery code).
Requires X-Recovery-Proof header containing the HMAC-SHA256 proof:
proof = HMAC-SHA256(key=enc_key_salt_bytes, msg=nonce_from_recovery_data)
Items are returned as encrypted ciphertext blobs only — no sensitive
plaintext is exposed. The client re-encrypts them locally.
The server validates the proof against the value precomputed and stored in
flask.session during /recovery/data — enc_key_salt is never sent in plaintext.
Items are returned as encrypted ciphertext blobs only.
"""
email = (request.args.get('email') or '').strip().lower()
recovery_proof = request.headers.get('X-Recovery-Proof', '').strip()
client_proof = request.headers.get('X-Recovery-Proof', '').strip()
if not email or not recovery_proof:
if not email or not client_proof:
return jsonify({'error': 'email and X-Recovery-Proof header are required'}), 400
# Validate session binding: proof must match what was issued to this session.
expected_proof = session.get('recovery_expected_proof', '')
session_user_id = session.get('recovery_user_id')
if not expected_proof or not session_user_id:
return jsonify({'error': 'No active recovery challenge. Call /recovery/data first.'}), 400
user = User.query.filter_by(email=email).first()
if not user or not user.recovery_enc_salt:
if not user or not user.recovery_enc_salt or user.id != session_user_id:
return jsonify({'error': 'No recovery data found'}), 404
if recovery_proof != user.enc_key_salt:
if not verify_recovery_proof(expected_proof, client_proof):
AuditLog.log(
user_id=user.id,
action='auth.recovery_items_denied',
@@ -689,5 +864,4 @@ def recovery_items():
{'id': item.id, 'enc_data': item.enc_data, 'iv': item.iv}
for item in items
]
}), 200
}), 200