Implement basic user roles functionality

api/functions/error_messages.py

@@ -30,6 +30,14 @@ def error_password_not_updated():
 	return str("Unable to update password, please try again")
 
 
+def error_role_not_updated():
+	return str("Unable to update user's role, please try again")
+
+
+def error_not_an_admin():
+	return str("You are not allowed to perform admin actions, please try again")
+
+
 def error_user_not_updated():
 	return str("User not updated, please try again")
 

api/functions/sign_in_user.py

@@ -24,8 +24,9 @@ def sign_in_user(email, password):
 	password_match = bcrypt.check_password_hash(user.user_password, password)
 
 	if email_match and password_match:
+		user_claims = {"roles": user.user_role}
 		temp_dict = {
-			'auth_token': create_access_token(user.id),
+			'auth_token': create_access_token(user.id, user_claims=user_claims),
 			'user': user
 		}
 

api/functions/update_user_role.py

@@ -0,0 +1,37 @@
+from functions.error_messages import *
+from db import db
+from models import Users as User
+from graphql import GraphQLError
+
+from flask_graphql_auth import *
+
+
+def update_user_role(email, new_role):
+    """
+        Updates the user role associate with the user given by email address
+
+        :param email: The email address associated with the user who's role will be updated.
+        :param new_role: The new role that will be given to the user.
+
+        :returns user: The newly updated user object retrieved from the DB (after the update is committed).
+    """
+    user = User.query.filter(User.user_email == email).first()
+
+    if user is None:
+        # State that no such user exists using that email address
+        raise GraphQLError(error_user_does_not_exist())
+
+    role = get_jwt_claims()['roles']  # Pulls the 'role' out of the JWT user claims associated with the token.
+
+    if role == "admin":  # If an admin, update the user.
+        user = User.query.filter(User.user_email == email)\
+                .update({'user_role': new_role})
+        db.session.commit()
+    else:
+        raise GraphQLError(error_not_an_admin())
+
+    if not user:
+        raise GraphQLError(error_role_not_updated())
+    else:
+        user = User.query.filter(User.user_email == email).first()
+        return user

api/manage.py

@@ -8,11 +8,12 @@
 	DB_NAME,
 	DB_HOST,
 	DB_PASS,
-	DB_USER
+	DB_USER,
+	DB_PORT,
 )
 
 app = Flask(__name__)
-app.config['SQLALCHEMY_DATABASE_URI'] = f'postgresql+psycopg2://{DB_USER}:{DB_PASS}@{DB_HOST}/{DB_NAME}'
+app.config['SQLALCHEMY_DATABASE_URI'] = f'postgresql+psycopg2://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}'
 app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
 

api/migrations/versions/5cd6b0bb4549_.py → api/migrations/versions/210111cbd5ae_.py

@@ -1,16 +1,16 @@
 """empty message
 
-Revision ID: 5cd6b0bb4549
+Revision ID: 210111cbd5ae
 Revises: 
-Create Date: 2020-02-07 13:12:26.620069
+Create Date: 2020-02-14 08:15:16.902318
 
 """
 from alembic import op
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
 # revision identifiers, used by Alembic.
-revision = '5cd6b0bb4549'
+revision = '210111cbd5ae'
 down_revision = None
 branch_labels = None
 depends_on = None
@@ -58,6 +58,8 @@ def upgrade():
     sa.Column('user_password', sa.String(), nullable=True),
     sa.Column('preferred_lang', sa.String(), nullable=True),
     sa.Column('failed_login_attempts', sa.Integer(), nullable=True),
+    sa.Column('tfa_validated', sa.Boolean(), nullable=True),
+    sa.Column('user_role', sa.String(), nullable=True),
     sa.PrimaryKeyConstraint('id')
     )
     op.create_table('admin_affiliations',

api/models.py

@@ -90,7 +90,8 @@ class Users(db.Model):
     user_password = Column(String)
     preferred_lang = Column(String)
     failed_login_attempts = Column(Integer, default=0)
-    two_factor_auth = Column(Boolean, default=False)
+    tfa_validated = Column(Boolean, default=False)
+    user_role = Column(String, default="user")
     user_affiliation = relationship("User_affiliations", back_populates="user", cascade="all, delete")
 
 

api/queries.py

@@ -1,5 +1,9 @@
 import os
 from graphene_sqlalchemy import SQLAlchemyConnectionField
+from graphene import String
+from flask_graphql_auth import *
+import pyotp
+
 import graphene
 from sqlalchemy.orm import joinedload
 from graphene import relay, String
@@ -24,6 +28,13 @@
 	resolve_get_group_by_id,
 	resolve_get_group_by_group,
 	resolve_get_group_by_sector
+
+)
+
+
+from resolvers.users import (
+	resolve_test_user_claims,
+	resolve_generate_otp_url,
 )
 
 
@@ -71,12 +82,17 @@ class Query(graphene.ObjectType):
 		description="Allows selection of groups from a given sector enum"
 	)
 
-	generate_otp_url = String(email=String(required=True))
+	generate_otp_url = graphene.String(
+		email=graphene.Argument(EmailAddress, required=True),
+		resolver=resolve_generate_otp_url,
+		description="An api endpoint used to generate a OTP url used for two factor authentication."
+	)
 
-	@staticmethod
-	def resolve_generate_otp_url(self, info, email):
-		totp = pyotp.totp.TOTP(os.getenv('BASE32_SECRET'))  # This needs to be a 16 char base32 secret key
-		return totp.provisioning_uri(email, issuer_name="Tracker")
+	test_user_claims = graphene.String(
+		token=graphene.Argument(graphene.String, required=True),
+		resolver=resolve_test_user_claims,
+		description="An api endpoint to view a current user's claims -- Requires an active JWT."
+	)
 
 
 class Mutation(graphene.ObjectType):
@@ -85,7 +101,7 @@ class Mutation(graphene.ObjectType):
 	sign_in = SignInUser.Field()
 	update_password = UpdateUserPassword.Field()
 	authenticate_two_factor = ValidateTwoFactor.Field()
+	update_user_role = UpdateUserRole.Field()
 
 
-schema = graphene.Schema(query=Query, mutation=Mutation)
-
+schema = graphene.Schema(query=Query, mutation=Mutation)

api/resolvers/users.py

@@ -0,0 +1,26 @@
+from flask_graphql_auth import *
+import pyotp
+import os
+
+
+@query_jwt_required
+def resolve_test_user_claims(self, info):
+    """
+        This resolver returns the user_claims array -- A utility for testing
+        It requires that a JWT token be active, and that the user have an admin role
+    """
+    role = get_jwt_claims()['roles']
+
+    if role == "admin":
+        return str(get_jwt_claims())
+    else:
+        return str("Not an admin, please log in")
+
+
+def resolve_generate_otp_url(self, info, email):
+    """
+        This resolver adds an api endpoint that returns a url for OTP validation
+        :param email: The email address that will be associated with the URL generated
+    """
+    totp = pyotp.totp.TOTP(os.getenv('BASE32_SECRET'))  # This needs to be a 16 char base32 secret key
+    return totp.provisioning_uri(email, issuer_name="Tracker")

api/schemas/user.py

@@ -1,11 +1,13 @@
 import graphene
 from graphene import relay
 from graphene_sqlalchemy import SQLAlchemyObjectType
+from flask_graphql_auth import *
 
 from functions.create_user import create_user
 from functions.sign_in_user import sign_in_user
 from functions.update_user_password import update_password
 from functions.validate_two_factor import validate_two_factor
+from functions.update_user_role import update_user_role
 
 from models import Users as User
 from scalars.email_address import *
@@ -69,6 +71,7 @@ def mutate(self, info, password, confirm_password, email):
 		user = update_password(email=email, password=password, confirm_password=confirm_password)
 		return UpdateUserPassword(user=user)
 
+
 class ValidateTwoFactor(graphene.Mutation):
 	class Arguments:
 		email = EmailAddress(required=True)
@@ -80,3 +83,17 @@ class Arguments:
 	def mutate(self, info, email, otp_code):
 		user_to_rtn = validate_two_factor(email=email, otp_code=otp_code)
 		return ValidateTwoFactor(user=user_to_rtn)
+
+
+class UpdateUserRole(graphene.Mutation):
+	class Arguments:
+		token = graphene.String(required=True)
+		email = EmailAddress(required=True)
+		role = graphene.String(required=True)
+
+	user = graphene.Field(lambda: UserObject)
+
+	@mutation_jwt_required
+	def mutate(self, info, email, role):
+		user = update_user_role(email=email, new_role=role)
+		return UpdateUserRole(user=user)