shopdb-flask/plugins/tools/api/routes.py

"""Secure send: create a one-view link for a credential, and redeem it.

Lives in Tech Tools because that is where somebody looks for a utility. It is
the one tool here that is NOT purely browser-side: a link that expires and dies
on first read needs somewhere to keep the ciphertext until then.

TWO DIFFERENT DOORS, on purpose.

CREATE needs a login and a permission. Anonymous creation would be an
untraceable way to move data out of a system that otherwise records who did
what, and the creator is the only person these rows can ever name.

READ needs neither, because the recipient usually has no account here and the
link is normally handed over in a call or a chat. The unguessable token IS the
authorisation - which is why it is 128 bits of randomness and why the row is
spent on the first successful read.

WHAT KEEPS THIS FROM BECOMING A DOCUMENT TRANSFER. The key never reaches the
server, so the server cannot look at what it holds and cannot tell a password
from a page of a drawing. The only lever it has is SIZE, and MAX_CIPHERTEXT is
therefore not a tuning knob - it is the whole enforcement. Everything else (line
counts, file signatures, refusing a pasted file) runs in the browser, guides an
honest person, and is bypassable by anyone willing to POST directly.
"""

import secrets as pysecrets
from datetime import datetime, timedelta, timezone

from flask import Blueprint, request
from flask_jwt_extended import jwt_required, get_jwt_identity

from shopdb.api import (
    db,
    success_response, error_response, ErrorCodes,
    require_permission,
)

from ..models import SecureSecret

tools_bp = Blueprint('tools', __name__)

# Base64 of an encrypted credential. A password is well under a hundred
# characters; this is generous for a connection string and useless for a
# document. See the module docstring: this limit is the only server-side control
# that a zero-knowledge design leaves available.
MAX_CIPHERTEXT = 4096

# Hours. The default is short because the normal case is "read it while we are
# on this call"; the maximum exists so nothing lingers for a week.
DEFAULT_EXPIRY_HOURS = 24
MAX_EXPIRY_HOURS = 168

MAX_VIEWS = 5


def _utcnow():
    return datetime.now(timezone.utc).replace(tzinfo=None)


@tools_bp.route('/secrets', methods=['POST'])
@jwt_required()
@require_permission('tools.secrets')
def create_secret():
    """Store an already-encrypted secret and return its token."""
    data = request.get_json(silent=True) or {}
    ciphertext = (data.get('ciphertext') or '').strip()
    iv = (data.get('iv') or '').strip()

    if not ciphertext or not iv:
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'ciphertext and iv are required')

    if len(ciphertext) > MAX_CIPHERTEXT:
        # Say what the limit is FOR, not just that it exists. Somebody hitting
        # it is usually trying to send a file, and the answer to that is a
        # different tool rather than a bigger number here.
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'Secret is too large ({} of {} characters). This is for passwords '
            'and short text, not documents - the server cannot read what it '
            'holds, so size is the only limit it can enforce.'.format(
                len(ciphertext), MAX_CIPHERTEXT))

    # `is None`, not `or`: an explicit 0 is falsy, so `or DEFAULT` silently
    # turned "expire immediately" into "expire in a day" instead of refusing it.
    # A caller who sends a number should be told when it is not allowed, not
    # quietly given a different one.
    rawhours = data.get('expiryhours')
    rawviews = data.get('views')
    try:
        hours = int(DEFAULT_EXPIRY_HOURS if rawhours is None else rawhours)
        views = int(1 if rawviews is None else rawviews)
    except (TypeError, ValueError):
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'expiryhours and views must be whole numbers')

    if not 1 <= hours <= MAX_EXPIRY_HOURS:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'expiryhours must be between 1 and {}'.format(MAX_EXPIRY_HOURS))
    if not 1 <= views <= MAX_VIEWS:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'views must be between 1 and {}'.format(MAX_VIEWS))

    try:
        userid = int(get_jwt_identity())
    except (TypeError, ValueError):
        userid = None

    secret = SecureSecret(
        token=pysecrets.token_hex(16),   # 128 bits; the link is the credential
        ciphertext=ciphertext,
        iv=iv,
        expiresat=_utcnow() + timedelta(hours=hours),
        viewsremaining=views,
        createdbyuserid=userid,
    )
    db.session.add(secret)
    db.session.commit()

    return success_response(secret.to_dict(), http_code=201)


@tools_bp.route('/secrets/<token>', methods=['POST'])
def redeem_secret(token):
    """Hand over the ciphertext once, then spend the view.

    POST rather than GET on purpose. A GET is prefetched by link previews in
    chat and mail clients, and a one-view secret consumed by a preview window is
    a secret the recipient never gets to see.

    A missing, expired and spent secret all answer the SAME way. Distinguishing
    them would confirm to anyone holding a guessed token whether it ever named
    anything.
    """
    gone = error_response(
        ErrorCodes.NOT_FOUND,
        'This link is not valid. It may have been opened already, or expired.',
        http_code=404)

    secret = SecureSecret.query.filter_by(token=(token or '').strip()).first()
    if secret is None:
        return gone

    now = _utcnow()
    if secret.is_expired(now) or secret.is_spent():
        # Tidy up on the way past rather than waiting for a sweeper: a spent
        # secret has no reason to keep its ciphertext.
        db.session.delete(secret)
        db.session.commit()
        return gone

    payload = {'ciphertext': secret.ciphertext, 'iv': secret.iv}

    secret.viewsremaining -= 1
    secret.viewcount += 1
    secret.lastviewedat = now
    if secret.viewsremaining <= 0:
        db.session.delete(secret)
    db.session.commit()

    return success_response(payload)


@tools_bp.route('/secrets/<token>', methods=['GET'])
def peek_secret(token):
    """Whether a link is still live, WITHOUT spending it.

    The read page asks this first so it can render "this secret is waiting, and
    opening it will destroy it" rather than consuming the only view to find out
    there was something there.
    """
    secret = SecureSecret.query.filter_by(token=(token or '').strip()).first()
    live = bool(secret and not secret.is_expired() and not secret.is_spent())
    return success_response({'available': live})