dtf-system/app/operators.py

"""Kanban operator accounts.

One shared login meant every card movement was attributed to the same name, so
the movement history could not answer who did what. Accounts live in the
database; `OPERATOR_EMAIL` and `OPERATOR_PASSWORD` seed the first one so an
existing deployment keeps working unchanged.

Administered from the API container, the same way the schema is:

    python -m local.operators list
    python -m local.operators add maria@example.com --name "Maria"
    python -m local.operators password maria@example.com
    python -m local.operators disable maria@example.com
    python -m local.operators enable maria@example.com

Passwords are read from the terminal, never from an argument, so they do not
reach shell history or the process list.
"""
import getpass
import os
import sys
from uuid import uuid4

from .core.auth import password_hash
from .core.db import connect

MIN_PASSWORD = 12


def seed_from_environment(cursor):
    """Make the configured credential a real account, once.

    Only inserts when that email is absent, so a password changed here is never
    reverted by a stale environment variable on the next deploy.
    """
    email = os.environ.get('OPERATOR_EMAIL', '').strip().lower()
    password = os.environ.get('OPERATOR_PASSWORD', '')
    if not email or not password:
        return None
    existing = cursor.execute(
        'SELECT id FROM dtf_local.operators WHERE email=%s', (email,)).fetchone()
    if existing:
        return None
    cursor.execute(
        'INSERT INTO dtf_local.operators(id,email,name,password_hash) VALUES(%s,%s,%s,%s)',
        (uuid4(), email, 'Operador', password_hash(password)))
    return email


def _ask_password(email):
    first = getpass.getpass(f'New password for {email}: ')
    if len(first) < MIN_PASSWORD:
        raise SystemExit(f'Password must be at least {MIN_PASSWORD} characters.')
    if first != getpass.getpass('Repeat: '):
        raise SystemExit('Passwords did not match.')
    return first


def add(email, name=''):
    email = email.strip().lower()
    with connect() as c:
        if c.execute('SELECT id FROM dtf_local.operators WHERE email=%s', (email,)).fetchone():
            raise SystemExit(f'{email} already exists. Use "password" or "enable".')
        c.execute('INSERT INTO dtf_local.operators(id,email,name,password_hash) VALUES(%s,%s,%s,%s)',
                  (uuid4(), email, name, password_hash(_ask_password(email))))
    print(f'Added {email}.')


def password(email):
    email = email.strip().lower()
    with connect() as c:
        if not c.execute('SELECT id FROM dtf_local.operators WHERE email=%s', (email,)).fetchone():
            raise SystemExit(f'{email} does not exist.')
        c.execute('UPDATE dtf_local.operators SET password_hash=%s WHERE email=%s',
                  (password_hash(_ask_password(email)), email))
    print(f'Password changed for {email}. Existing sessions stay valid until they expire;'
          ' run "disable" first if the account is compromised.')


def set_active(email, active):
    email = email.strip().lower()
    with connect() as c:
        updated = c.execute(
            'UPDATE dtf_local.operators SET active=%s WHERE email=%s RETURNING email',
            (active, email)).fetchone()
        if not updated:
            raise SystemExit(f'{email} does not exist.')
        if not active:
            # Revoke immediately: disabling must end access now, not in eight hours.
            c.execute('DELETE FROM dtf_local.operator_sessions WHERE username=%s', (email,))
    print(f'{email} {"enabled" if active else "disabled; open sessions revoked"}.')


def listing():
    with connect() as c:
        rows = c.execute('''SELECT email,name,active,last_login_at FROM dtf_local.operators
                            ORDER BY email''').fetchall()
    if not rows:
        print('No operator accounts. Set OPERATOR_EMAIL and OPERATOR_PASSWORD and run'
              ' "python -m local.bootstrap", or add one here.')
        return
    for row in rows:
        seen = row['last_login_at'].strftime('%Y-%m-%d %H:%M') if row['last_login_at'] else 'never'
        state = 'active ' if row['active'] else 'DISABLED'
        print(f"{state}  {row['email']:<34} {row['name'][:20]:<20} last login {seen}")


def main(argv):
    if not argv:
        raise SystemExit(__doc__)
    command, rest = argv[0], argv[1:]
    if command == 'list':
        return listing()
    if not rest:
        raise SystemExit(f'usage: python -m local.operators {command} <email>')
    email = rest[0]
    if command == 'add':
        name = rest[rest.index('--name') + 1] if '--name' in rest else ''
        return add(email, name)
    if command == 'password':
        return password(email)
    if command == 'disable':
        return set_active(email, False)
    if command == 'enable':
        return set_active(email, True)
    raise SystemExit(__doc__)


if __name__ == '__main__':
    main(sys.argv[1:])