Add invitations

roc_fnb/website/server/server.py

@@ -1,16 +1,17 @@
 from functools import wraps
 import json
+from os import environ
 from pathlib import Path
 from random import randbytes
 from sys import stderr
 
-from flask import Flask, redirect
+from flask import Flask, redirect, session, g
 
 from roc_fnb.util.env_file import env_file
 from roc_fnb.util import log
 from roc_fnb.website.database import Database
 from roc_fnb.website.server.user import setup_user_routes
-
+from roc_fnb.website.models.user import JwtUser
 
 db = Database.from_env()
 
@@ -21,13 +22,20 @@ app = Flask(
     static_folder=Path(__file__).absolute().parent / 'static',
 )
 
-app.secret_key = env_file('FLASK_SECRET', default_file='./flask.secret', default_fn=lambda: randbytes(12))
+app.secret_key = env_file(
+    'FLASK_SECRET',
+    default_file='./flask.secret',
+    default_fn=lambda: randbytes(12)
+)
+APP_HOSTNAME = environ["APP_HOSTNAME"]
+
 
 @app.before_request
 def decode_user():
     if user := session.get('user'):
         g.user = JwtUser.from_json(data=json.loads(user))
-        
+    g.app_hostname = APP_HOSTNAME
+
 
 @app.route('/ig')
 def ig_redir():
@@ -43,4 +51,5 @@ def donate_redir():
 def index():
     return redirect('/index.html')
 
-setup_user_routes(app, db)
+
+setup_user_routes(app, db)

roc_fnb/website/server/test_user.py

@@ -0,0 +1,129 @@
+from bson.objectid import ObjectId
+from flask.testing import FlaskClient
+import json
+from pytest import fixture
+
+from roc_fnb.util import base64_decode
+from roc_fnb.website.server import app
+from roc_fnb.website.models.user import User
+from roc_fnb.website.database import Database
+
+
+@fixture
+def user() -> User:
+    return User.create('test@t.co', 'name', 'monkey')
+
+
+@fixture
+def database() -> Database:
+    return Database.from_env()
+
+
+@fixture
+def client():
+    return app.test_client()
+
+
+def test_login(user: User, database: Database, client: FlaskClient):
+    database.store_user(user)
+    try:
+        assert user._id is not None  # for mypy
+        resp = client.post(
+            '/login', json={
+                'name': user.name,
+                'password': 'monkey'
+            }
+        )
+        assert resp.json is not None  # for mypy
+        assert resp.json['status'] == "OK"
+        with client.session_transaction() as session:
+            session_data = json.loads(session['user'])
+        assert base64_decode(session_data['_id']) == user._id.binary
+        assert session_data['name'] == user.name
+        assert 'password' not in session_data.keys()
+        assert session_data['email'] == user.email
+        assert not session.get('admin')
+        assert not session.get('moderator')
+    finally:
+        if _id := user._id:
+            database.delete_user(_id)
+
+
+def test_create_user(user: User, database: Database, client: FlaskClient):
+    user.admin = True
+    try:
+        database.store_user(user)
+        assert user._id is not None  # for mypy
+        resp = client.post(
+            '/login', json={
+                'name': user.name,
+                'password': 'monkey'
+            }
+        )
+        assert resp.json is not None  # for mypy
+        assert resp.json['status'] == "OK"
+        with client.session_transaction() as session:
+            assert session.get('user') is not None
+            session_data = json.loads(session['user'])
+            assert session_data['admin']
+        resp = client.get('/invite', headers={'Accept': 'application/json'})
+        assert resp.status_code == 200
+        assert resp.json is not None
+        invite = resp.json['invite']
+
+        with client.session_transaction() as session:
+            del session['user']
+        _id = None
+        try:
+            resp = client.post(
+                '/new-user',
+                json={
+                    'name': 'Test 2',
+                    'password': 'hunter2',
+                    'invite_code': invite,
+                    'email': None
+                }
+            )
+            assert resp.status_code == 200
+            assert resp.json is not None
+            assert resp.json['status'] == 'OK'
+            with client.session_transaction() as session:
+                session_data = json.loads(session['user'])
+            _id = ObjectId(base64_decode(session_data['_id']))
+            new_user = database.get_user_by_id(_id)
+            assert new_user is not None
+            assert new_user._id == _id
+            assert new_user.name == 'Test 2'
+            assert new_user.check_password('hunter2')
+            assert not (new_user.admin or new_user.moderator)
+            invitation_doc = database.db.invitations.find_one({'invite_code': base64_decode(invite)})
+            assert invitation_doc is not None
+            assert invitation_doc['invited_user_id'] == _id
+            assert invitation_doc['invited_by'] == user._id
+        finally:
+            if _id is not None:
+                database.delete_user(_id)
+    finally:
+        if u_id := user._id:
+            database.delete_user(u_id)
+
+
+def test_create_user_deny_non_admin(user: User, database: Database, client: FlaskClient):
+    database.store_user(user)
+    try:
+        assert user._id is not None  # for mypy
+        resp = client.post(
+            '/login', json={
+                'name': user.name,
+                'password': 'monkey'
+            }
+        )
+        assert resp.json is not None  # for mypy
+        assert resp.json['status'] == "OK"
+        with client.session_transaction() as session:
+            assert session.get('user') is not None
+        resp = client.get('/invite', headers={'Accept': 'application/json'})
+        assert resp.status_code == 401
+    finally:
+        if _id := user._id:
+            database.delete_user(_id)

roc_fnb/website/server/user.py

@@ -1,9 +1,12 @@
 import json
 
-from flask import request, redirect, render_template, g, abort
+from flask import request, redirect, render_template, g, abort, make_response, flash, jsonify, session
 
-from roc_fnb.util import log
+from roc_fnb.util import log, base64_encode, base64_decode
 from roc_fnb.website.server.decorators import require_user, logger_request_bindings
+from roc_fnb.website.database import InvitationClaimed, InvitationNotFound
+from roc_fnb.website.models.user import User
+
 
 def setup_user_routes(app, db):
     @app.post('/login')
@@ -16,7 +19,7 @@ def setup_user_routes(app, db):
             log.warn('incorrect password submitted', name=form['name'])
             abort(401)  # unauthorized
         session['user'] = json.dumps(user.public_fields)
-        return redirect('/me')
+        return jsonify(status='OK')
 
     @app.get('/login')
     def render_login_page():
@@ -28,4 +31,72 @@ def setup_user_routes(app, db):
     @app.get('/me')
     @require_user()
     def get_profile():
-        return render_template('profile.html', user=g.user)
+        return render_template('profile.html', user=g.user)
+
+    @app.get('/invite')
+    @require_user(admin=True)
+    def create_invite():
+        """
+        Two handlers in one: JSON and HTML.
+        
+        First, a user-agent (browser) makes a request for /invite, and gets the
+        rendered HTML page. Then they click a button which sends a request
+        specifically asking for a JSON reply. The invitation is created and
+        returned in a JSON document.
+
+        This allows a user to generate more than one invitation code per visit
+        to the page and avoids accidentally creating an invite code on page load.
+        """
+        if request.headers['Accept'] == 'application/json':
+            invite = base64_encode(db.invite_new_user(g.user))
+            log.info('new invitation created', inviter=g.user, invitation_code=invite)
+            return jsonify(invite=invite, status='OK')
+        return render_template(
+            'new_invite.html', user=g.user, app_hostname=g.app_hostname
+        )
+
+    @app.post('/new-user')
+    def create_new_user():
+        decoded_invite_code = base64_decode(request.json['invite_code'])
+        invitee = User.create(
+            request.json['email'], request.json['name'],
+            request.json['password']
+        )
+        try:
+            db.store_new_invitee(decoded_invite_code, invitee)
+            log.info('new user created', user=invitee, invite_code=request.json['invite_code'])
+        except InvitationClaimed as err:
+            response = make_response(
+                json.dumps(
+                    {
+                        'type':
+                            'InvitationClaimed',
+                        'invite_code':
+                            base64_encode(err.invitation['invite_code']),
+                        'message':
+                            str(err)
+                    }
+                )
+            )
+            response.headers['Content-Type'] = 'application/json'
+            response.status_code = 401
+            abort(response)
+        except InvitationNotFound as err:
+            r = make_response(
+                json.dumps(
+                    {
+                        'type': 'InvitationNotFound',
+                        'invite_code': err.invitation_code,
+                        'message': str(err)
+                    }
+                )
+            )
+            response.headers['Content-Type'] = 'application/json'
+            response.status_code = 404
+            abort(response)
+        session['user'] = json.dumps(invitee.public_fields)
+        return jsonify(status='OK')
+
+    @app.get('/new-user')
+    def render_signup_page():
+        return render_template('sign-up.html')