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 nothing more. Sharing a password is something anybody
at the site may legitimately need to do, and a permission somebody has to be
granted first turns a two-minute job into a ticket - which is how people end up
pasting credentials into chat instead.

The login stays, though, and is not a formality: anonymous creation would make
this an unattributable write endpoint that stores arbitrary ciphertext, and the
creator is the only person these rows can ever name. Everybody browsing this app
is already signed in, so requiring it costs nothing.

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 import current_app
from flask_jwt_extended import jwt_required, get_jwt_identity, current_user

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

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

# Generous, because "how many people need this password" is the site's business
# and not a number worth arguing with. The floor matters more than the ceiling:
# a secret readable an unlimited number of times is a password on a web page.
MAX_VIEWS = 100

# Guest creation, when a site enables it, is an unauthenticated write. Bounded
# per address so it cannot be used to fill a table or park data: a person
# sharing passwords does it a handful of times an hour, not hundreds.
GUEST_CREATE_MAX = 10
GUEST_CREATE_WINDOW_SECONDS = 600


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


def _sweep_expired():
    """Delete secrets whose window has passed, whoever asked.

    Redeeming removes a row, and so does touching an expired one - but a secret
    nobody ever opens was removed by NOTHING. The recipient goes on holiday, the
    link is never clicked, and the ciphertext sits there for good: not a
    disclosure risk, since it cannot be read, but it makes "nothing lingers"
    untrue and grows the table in exactly the case where the feature failed.

    Swept opportunistically rather than on a schedule: creating a secret is
    authenticated and infrequent, and an air-gapped site is not going to have a
    cron job anybody remembers to add.
    """
    stale = SecureSecret.query.filter(SecureSecret.expiresat < _utcnow()).all()
    for secret in stale:
        db.session.delete(secret)
    return len(stale)


def _guests_allowed():
    """May somebody with no account create a secret? Default NO.

    Read straight from the row rather than the settings cache: this decides
    whether an unauthenticated caller may write, and being wrong on the
    permissive side for five minutes is the wrong way round to be wrong.
    """
    try:
        row = Setting.query.filter_by(key='securesend_allow_guests').first()
    except Exception:
        # FAIL CLOSED. If the setting cannot be read - the row is missing, the
        # table is not there yet on a half-migrated site, the database is
        # unhappy - the answer is no. An error must never be the thing that
        # opens an unauthenticated write endpoint.
        current_app.logger.exception('could not read securesend_allow_guests')
        return False
    return bool(row and str(row.value).strip().lower() in ('true', '1', 'yes'))


def _guest_ratelimited():
    """Bounded guest creation, per address, in a fixed window.

    Same shape as the login limiter. Not a security boundary - a determined
    person has more addresses - but it stops an open endpoint being an easy
    place to park data or fill a disk.
    """
    address = (request.headers.get('X-Forwarded-For', '').split(',')[0].strip()
               or request.remote_addr or 'unknown')
    bucket = int(_utcnow().timestamp()) // GUEST_CREATE_WINDOW_SECONDS
    key = f'securesendguest:{address}:{bucket}'
    used = cache.get(key) or 0
    if used >= GUEST_CREATE_MAX:
        return True
    cache.set(key, used + 1, timeout=GUEST_CREATE_WINDOW_SECONDS)
    return False


@tools_bp.route('/secrets', methods=['POST'])
@jwt_required(optional=True)
def create_secret():
    """Store an already-encrypted secret and return its token.

    OPTIONAL jwt so the SETTING decides rather than the decorator: a guest has
    to reach the handler to be told whether this site lets them in.
    """
    # WHO MAY CREATE. Signed in: the tools.secrets permission, with the usual
    # admin bypass. Not signed in: only when the site has deliberately opened
    # it, and then within a rate limit.
    # get_jwt_identity(), NOT `current_user is None`: current_user is a
    # LocalProxy that is never None itself, so that test silently took the
    # signed-in branch for a guest and then blew up dereferencing it.
    identity = get_jwt_identity()
    if identity is None:
        if not _guests_allowed():
            return error_response(
                ErrorCodes.UNAUTHORIZED,
                'Sign in to create a secret link.',
                http_code=401)
        if _guest_ratelimited():
            return error_response(
                ErrorCodes.VALIDATION_ERROR,
                'Too many links from this address. Try again shortly.',
                http_code=429)
    elif not current_user.haspermission('tools.secrets'):
        return error_response(
            ErrorCodes.FORBIDDEN,
            'You do not have permission to create secret links.',
            http_code=403)
    # Tidy up on the way past. Cheap, and it keeps the table bounded without a
    # scheduler existing anywhere.
    _sweep_expired()

    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())
    if secret is not None and not live:
        # A late click tidies its own row, so a dead secret does not need
        # somebody to create a new one before it is cleaned up.
        db.session.delete(secret)
        db.session.commit()
    return success_response({'available': live})