agilentics / boiler 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
"""Sign in, sign out, and which organisation the session is looking at.

The worked example of a façade: each endpoint reads the request, calls one core
function, and serialises the result. No rule about who may do what is decided
here - that lives in app/core.
"""
import logging

from flask import jsonify, redirect, render_template, request
from flask import session as flask_session

from app.core import auth as auth_core
from app.core import orgs as orgs_core
from db.db_declarations import User
from utils.db_wrapper import db_session

from .crud import ORG_KEY, USER_KEY, endpoint, principal

logger = logging.getLogger(__name__)


def _begin_session(user_id: int, org_id: int | None) -> None:
    """Start an authenticated session. Every sign-in path funnels through here.

    `clear()` before anything is written: a pre-existing session id must not
    survive a successful sign-in, or an attacker who fixed the cookie beforehand
    holds a session that is now authenticated as the victim. It also drops the
    CSRF token, so the next rendered page mints a fresh one.
    """
    flask_session.clear()
    flask_session[USER_KEY] = user_id
    flask_session[ORG_KEY] = org_id
    flask_session.permanent = True


def page_login():
    """The sign-in page. Already signed in, it is not a form but a redirect -
    showing a login form to someone who is logged in invites them to believe
    their session ended."""
    if flask_session.get(USER_KEY):
        return redirect("/")
    return render_template("login.html")


@endpoint
def api_login():
    """Exchange an email and password for a session."""
    body = request.get_json(silent=True) or {}
    user = auth_core.authenticate_password(
        db_session, body.get("email", ""), body.get("password", ""))

    # Resolve the principal before starting the session, so an account with no
    # membership is refused rather than left holding a session it cannot use.
    p = auth_core.principal_for_user(db_session, user)

    _begin_session(user.id, p.org_id)

    logger.info("Signed in %s into org %s.", user.email, p.org_id)
    return jsonify({"user": user.to_dict(), "org_id": p.org_id})


def page_logout():
    flask_session.clear()
    return redirect("/login")


@endpoint
def api_me():
    """Who the current session is, and which organisations it may enter - what
    the account menu and the org switcher render from."""
    p = principal()
    user = db_session.get(User, p.user_id)
    return jsonify({
        "user": user.to_dict(),
        "org_id": p.org_id,
        "scopes": sorted(p.scopes),
        "orgs": [o.to_dict() for o in orgs_core.orgs_for_user(db_session, p.user_id)],
    })


@endpoint
def api_org_select():
    """Point the session at another of this account's organisations.

    Validated through orgs_for_user rather than trusted from the body: the org id
    in the session is what every later query is scoped by, so accepting one the
    caller does not belong to would hand them another tenant's data for the rest
    of the session.
    """
    p = principal()
    body = request.get_json(silent=True) or {}
    org_id = body.get("org_id")

    if not any(o.id == org_id for o in orgs_core.orgs_for_user(db_session, p.user_id)):
        # NotFound, not Forbidden - see orgs.get_org.
        return jsonify({"error": "No such organisation."}), 404

    flask_session[ORG_KEY] = org_id
    return jsonify({"org_id": org_id})


@endpoint
def api_password_change():
    p = principal()
    body = request.get_json(silent=True) or {}
    auth_core.change_password(
        db_session, p, body.get("current", ""), body.get("new", ""))
    return jsonify({"ok": True})