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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
"""Administrative commands that need to run before anyone can sign in.

    python manage.py bootstrap --email you@example.com   # first org + owner
    python manage.py adduser  --email dev@example.com    # add a user to an org
    python manage.py passwd   --email you@example.com    # reset a password

`bootstrap` exists because authentication has a chicken-and-egg: the app needs an
account to sign in with, and creating one needs the app. It is a command rather
than a migration - a user created by a migration is one whose password nobody
chose, sitting in every deployment forever.
"""
import argparse
import getpass
import logging
import sys

from app.core import CoreError, Principal
from app.core import auth as auth_core
from app.core import orgs as orgs_core
from db.db_declarations import Organization, User
from settings import Config
from utils.db_wrapper import SessionLocal

logging.basicConfig(level=logging.INFO, format="%(levelname)s:%(message)s")


def _read_password(args=None, prompt="Password (min 12 chars): ") -> str:
    """Read a password without echoing it. --password-stdin reads a single line
    for the case where there is no terminal (CI, a container, an agent running
    this on someone's behalf); piping is the only non-interactive option, because
    a --password flag would put the secret in the process list."""
    if args is not None and getattr(args, "password_stdin", False):
        # Two invisible, fatal transport artefacts: a BOM PowerShell prepends
        # when piping to a native exe, and a Windows \r. Either produces an
        # account that is set successfully and can never be signed into.
        password = sys.stdin.readline().lstrip("").rstrip("\r\n")
        if not password:
            raise SystemExit("No password on stdin.")
        return password

    first = getpass.getpass(prompt)
    if first != getpass.getpass("Again: "):
        raise SystemExit("Those did not match.")
    return first


def bootstrap(args) -> None:
    session = SessionLocal()
    try:
        org = session.query(Organization).filter(
            Organization.slug == args.org).one_or_none()
        if org is None:
            org = orgs_core.create_org(session, Principal.system(),
                                       name=args.org_name or args.org.title(),
                                       slug=args.org)
            print(f"Created organisation {org.slug!r}.")
        else:
            print(f"Using existing organisation {org.slug!r}.")

        user = session.query(User).filter(User.email == args.email).one_or_none()
        if user is None:
            user = auth_core.create_user(session, args.email,
                                         _read_password(args), args.name)
            print(f"Created account {user.email!r}.")
        else:
            print(f"Using existing account {user.email!r}.")

        auth_core.add_membership(session, user, org, "owner")
        print(f"{user.email} is now an owner of {org.slug}.")
    finally:
        session.close()


def adduser(args) -> None:
    session = SessionLocal()
    try:
        org = session.query(Organization).filter(
            Organization.slug == args.org).one_or_none()
        if org is None:
            raise SystemExit(f"No organisation with slug {args.org!r}. "
                             "Run `bootstrap` first.")
        user = session.query(User).filter(User.email == args.email).one_or_none()
        if user is None:
            user = auth_core.create_user(session, args.email,
                                         _read_password(args), args.name)
        auth_core.add_membership(session, user, org, args.role)
        print(f"{user.email} is a {args.role} of {org.slug}.")
    finally:
        session.close()


def passwd(args) -> None:
    session = SessionLocal()
    try:
        user = session.query(User).filter(User.email == args.email).one_or_none()
        if user is None:
            raise SystemExit(f"No account for {args.email!r}.")
        user.password_hash = auth_core.hash_password(_read_password(args))
        # Clear the lockout too: an operator resetting a password is the
        # intended way out of one.
        user.failed_logins = 0
        user.locked_until = None
        session.commit()
        print(f"Password set for {user.email}.")
    finally:
        session.close()


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)

    def common(p):
        p.add_argument("--email", required=True)
        p.add_argument("--name")
        p.add_argument("--org", default=Config.DEFAULT_ORG_SLUG)
        p.add_argument("--password-stdin", action="store_true",
                       help="read the password from stdin instead of prompting")

    b = sub.add_parser("bootstrap", help="create the first organisation and owner")
    common(b)
    b.add_argument("--org-name")
    b.set_defaults(func=bootstrap)

    a = sub.add_parser("adduser", help="add a user to an organisation")
    common(a)
    a.add_argument("--role", default="member", choices=auth_core.ROLES)
    a.set_defaults(func=adduser)

    p = sub.add_parser("passwd", help="set an account's password")
    common(p)
    p.set_defaults(func=passwd)

    args = parser.parse_args()
    try:
        args.func(args)
    except CoreError as e:
        raise SystemExit(str(e))


if __name__ == "__main__":
    main()