Merge pull request #53 from nicholasaleks/feature/auth

Feature/auth

README.md

@@ -60,6 +60,7 @@ DVGA supports Beginner and Expert level game modes, which will change the exploi
   * HTML Injection
   * SQL Injection
 * **Authorization Bypass**
+  * GraphQL JWT Token Forge
   * GraphQL Interface Protection Bypass
   * GraphQL Query Deny List Bypass
 * **Miscellaneous**

app.py

@@ -5,13 +5,19 @@
 from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
 from flask_sockets import Sockets
+from flask_graphql_auth import GraphQLAuth
 
 app = Flask(__name__, static_folder="static/")
 app.secret_key = os.urandom(24)
 app.config["SQLALCHEMY_DATABASE_URI"] = config.SQLALCHEMY_DATABASE_URI
 app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = config.SQLALCHEMY_TRACK_MODIFICATIONS
 app.config["UPLOAD_FOLDER"] = config.WEB_UPLOADDIR
+app.config['SECRET_KEY'] = 'dvga'
+app.config["JWT_SECRET_KEY"] = 'dvga'
+app.config["JWT_ACCESS_TOKEN_EXPIRES"] = 120
+app.config["JWT_REFRESH_TOKEN_EXPIRES"] = 30
 
+auth = GraphQLAuth(app)
 sockets = Sockets(app)
 
 app.app_protocol = lambda environ_path_info: 'graphql-ws'
@@ -30,4 +36,4 @@
 
   server = pywsgi.WSGIServer((config.WEB_HOST, int(config.WEB_PORT)), app, handler_class=WebSocketHandler)
   print("DVGA Server Version: {version} Running...".format(version=VERSION))
-  server.serve_forever()
+  server.serve_forever()

core/helpers.py

@@ -1,9 +1,8 @@
-from config import WEB_UPLOADDIR
-from flask import session
 import base64
 import uuid
 import os
-
+from config import WEB_UPLOADDIR
+from jwt import decode
 from core.models import ServerMode
 
 def run_cmd(cmd):
@@ -18,6 +17,9 @@ def generate_uuid():
 def decode_base64(text):
   return base64.b64decode(text).decode('utf-8')
 
+def get_identity(token):
+  return decode(token, options={"verify_signature":False}).get('identity')
+
 def save_file(filename, text):
   try:
     f = open(WEB_UPLOADDIR + filename, 'w')

core/models.py

@@ -1,6 +1,7 @@
 import datetime
 
 from app import db
+import re
 from graphql import parse
 from graphql.execution.base import ResolveInfo
 
@@ -9,6 +10,7 @@ class User(db.Model):
     __tablename__ = 'users'
     id = db.Column(db.Integer, primary_key=True)
     username = db.Column(db.String(20),unique=True,nullable=False)
+    email = db.Column(db.String(20),unique=True,nullable=False)
     password = db.Column(db.String(60),nullable=False)
 
     @classmethod
@@ -19,6 +21,13 @@ def create_user(cls, **kw):
 
       return obj
 
+
+def clean_query(gql_query):
+  clean = re.sub(r'(?<=token:")(.*)(?=")', "*****", gql_query)
+  clean = re.sub(r'(?<=password:")(.*)(?=")', "*****", clean)
+  return clean
+
+
 class Audit(db.Model):
   __tablename__ = 'audits'
   id = db.Column(db.Integer, primary_key=True)
@@ -59,11 +68,13 @@ def create_audit_entry(cls, info, subscription_type=False):
           """Array-based Batch"""
           for i in info.context.json:
             gql_query = i.get("query")
+            gql_query = clean_query(gql_query)
             obj = cls(**{"gqloperation":gql_operation, "gqlquery":gql_query})
             db.session.add(obj)
         else:
           if info.context.json:
             gql_query = info.context.json.get("query")
+            gql_query = clean_query(gql_query)
             obj = cls(**{"gqloperation":gql_operation, "gqlquery":gql_query})
             db.session.add(obj)
 

core/security.py

@@ -66,7 +66,7 @@ def on_denylist(query):
   return False
 
 def operation_name_allowed(operation_name):
-  opnames_allowed = ['CreatePaste', 'EditPaste', 'getPastes', 'UploadPaste', 'ImportPaste']
+  opnames_allowed = ['CreatePaste', 'CreateUser', 'EditPaste', 'getPastes', 'UploadPaste', 'ImportPaste']
   if operation_name in opnames_allowed:
     return True
   return False

core/views.py

@@ -5,6 +5,7 @@
   helpers,
   middleware
 )
+from graphql.error import GraphQLError
 from core.directives import *
 from core.models import (
   Owner,
@@ -25,11 +26,19 @@
   make_response,
   session
 )
+
+from flask_graphql_auth import (
+    get_jwt_identity,
+    create_access_token,
+    create_refresh_token,
+)
+
+from flask_graphql_auth.decorators import verify_jwt_in_argument
 from flask_sockets import Sockets
 from graphql.backend import GraphQLCoreBackend
 from sqlalchemy import event, text
 from graphene_sqlalchemy import SQLAlchemyObjectType
-
+from core.helpers import get_identity
 from app import app, db
 
 from version import VERSION
@@ -39,7 +48,7 @@
 class UserObject(SQLAlchemyObjectType):
   class Meta:
     model = User
-    exclude_fields = ('password',)
+    exclude_fields = ('email',)
 
   username = graphene.String(capitalize=graphene.Boolean())
 
@@ -49,6 +58,13 @@ def resolve_username(self, info, **kwargs):
       return self.username.capitalize()
     return self.username
 
+  @staticmethod
+  def resolve_password(self, info, **kwargs):
+    if info.context.json.get('identity') == 'admin':
+      return self.password
+    else:
+      return '******'
+
 class PasteObject(SQLAlchemyObjectType):
   class Meta:
     model = Paste
@@ -71,6 +87,7 @@ class Meta:
 
 class UserInput(graphene.InputObjectType):
   username = graphene.String(required=True)
+  email = graphene.String(required=True)
   password = graphene.String(required=True)
 
 class CreateUser(graphene.Mutation):
@@ -82,6 +99,7 @@ class Arguments:
   def mutate(root, info, user_data=None):
     user_obj = User.create_user(
       username=user_data.username,
+      email=user_data.email,
       password=user_data.password
     )
 
@@ -205,13 +223,32 @@ def mutate(self, info, host='pastebin.com', port=443, path='/', scheme="http"):
 
     return ImportPaste(result=cmd)
 
+class Login(graphene.Mutation):
+    access_token = graphene.String()
+    refresh_token = graphene.String()
+
+    class Arguments:
+        username = graphene.String()
+        password = graphene.String()
+
+    def mutate(self, info , username, password) :
+        user = User.query.filter_by(username=username, password=password).first()
+        Audit.create_audit_entry(info)
+        if not user:
+            raise Exception('Authentication Failure')
+        return Login(
+            access_token = create_access_token(username),
+            refresh_token = create_refresh_token(username)
+        )
+
 class Mutations(graphene.ObjectType):
   create_paste = CreatePaste.Field()
   edit_paste = EditPaste.Field()
   delete_paste = DeletePaste.Field()
   upload_paste = UploadPaste.Field()
   import_paste = ImportPaste.Field()
   create_user = CreateUser.Field()
+  login = Login.Field()
 
 global_event = Subject()
 
@@ -229,6 +266,7 @@ class SearchResult(graphene.Union):
   class Meta:
     types = (PasteObject, UserObject)
 
+
 class Query(graphene.ObjectType):
   pastes = graphene.List(PasteObject, public=graphene.Boolean(), limit=graphene.Int(), filter=graphene.String())
   paste = graphene.Field(PasteObject, id=graphene.Int(), title=graphene.String())
@@ -241,6 +279,19 @@ class Query(graphene.ObjectType):
   search = graphene.List(SearchResult, keyword=graphene.String())
   audits = graphene.List(AuditObject)
   delete_all_pastes = graphene.Boolean()
+  me = graphene.Field(UserObject, token=graphene.String())
+
+  def resolve_me(self, info, token):
+    Audit.create_audit_entry(info)
+
+    identity = get_identity(token)
+
+    info.context.json['identity'] = identity
+
+    query = UserObject.get_query(info)
+
+    result = query.filter_by(username=identity).first()
+    return result
 
   def resolve_search(self, info, keyword=None):
     Audit.create_audit_entry(info)

db/solutions.py

@@ -20,4 +20,5 @@
     "partials/solutions/solution_19.html",
     "partials/solutions/solution_20.html",
     "partials/solutions/solution_21.html",
+    "partials/solutions/solution_22.html",
 ]

requirements.txt

@@ -27,7 +27,7 @@ packaging==21.3
 pluggy==1.0.0
 promise==2.3
 py==1.11.0
-PyJWT==2.3.0
+PyJWT==1.7.1
 pyparsing==3.0.9
 Pypubsub==4.0.3
 pytest==7.0.1

setup.py

@@ -58,8 +58,10 @@ def pump_db():
   print('Populating Database')
   db.create_all()
 
-  admin = User(username="admin", password=random_password())
-  operator = User(username="operator", password=random_password())
+  admin = User(username="admin", email="[email protected]", password=random_password())
+  operator = User(username="operator", email="[email protected]", password="password123")
+  # create tokens for admin & operator
+
   db.session.add(admin)
   db.session.add(operator)
 

templates/partials/solutions/solution_22.html

@@ -0,0 +1,21 @@
+<!-- Start -->
+<h3 style="color:purple" id="bypassauthz-token"><b>Authorization Bypass :: GraphQL JWT Token Forge</b></h3>
+<hr />
+<h5>Problem Statement</h5>
+<p>
+  Without logging in a user is able to forge the user identity claim within the JWT token for the <code>me</code> query operation.
+</p>
+<h5>Exploitation Solution <button class="reveal" onclick="reveal('sol-brokenauthz-token')">Show</button></h5>
+<div id="sol-brokenauthz-token" style="display:none">
+  <pre class="bash">
+
+query {
+    me(token: "FORGED_TOKEN") {
+        id
+        username
+        password
+    }
+}
+</pre>
+</div>
+<!-- End -->

templates/solutions.html

@@ -105,6 +105,9 @@ <h1 class="mt-4">Challenge Solutions</h1>
                     <li>
                       <b>Authorization Bypass</b>
                       <ul>
+                        <li>
+                          <a href="#bypassauthz-token">GraphQL JWT Token Forge</a>
+                        </li>
                         <li>
                           <a href="#bypassauthz-igql">GraphQL Interface Protection Bypass</a>
                         </li>

tests/test_auth.py

@@ -0,0 +1,64 @@
+import requests
+
+from common import GRAPHQL_URL, graph_query
+
+def test_mutation_login_success():
+  query = '''
+  mutation {
+    login(username: "operator", password:"password123") {
+      accessToken
+    }
+  }
+  '''
+  r = graph_query(GRAPHQL_URL, query)
+
+  assert r.json()['data']['login']['accessToken']
+
+
+def test_mutation_login_error():
+  query = '''
+  mutation {
+    login(username: "operator", password:"dolevwashere") {
+      accessToken
+    }
+  }
+  '''
+  r = graph_query(GRAPHQL_URL, query)
+
+  assert r.json()['errors'][0]['message'] == 'Authentication Failure'
+
+
+def test_query_me():
+  query = '''
+  query {
+    me(token: "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ0eXBlIjoiYWNjZXNzIiwiaWF0IjoxNjU2ODE0OTQ4LCJuYmYiOjE2NTY4MTQ5NDgsImp0aSI6ImI5N2FmY2QwLTUzMjctNGFmNi04YTM3LTRlMjdjODY5MGE2YyIsImlkZW50aXR5IjoiYWRtaW4iLCJleHAiOjE2NTY4MjIxNDh9.-56ZQN9jikpuuhpjHjy3vLvdwbtySs0mbdaSq-9RVGg") {
+      id
+      username
+      password
+    }
+  }
+  '''
+
+  r = graph_query(GRAPHQL_URL, query)
+
+  assert r.json()['data']['me']['id'] == '1'
+  assert r.json()['data']['me']['username'] == 'admin'
+  assert r.json()['data']['me']['password'] == 'changeme'
+
+
+def test_query_me_operator():
+  query = '''
+  query {
+    me(token: "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ0eXBlIjoiYWNjZXNzIiwiaWF0IjoxNjU2ODE0OTQ4LCJuYmYiOjE2NTY4MTQ5NDgsImp0aSI6ImI5N2FmY2QwLTUzMjctNGFmNi04YTM3LTRlMjdjODY5MGE2YyIsImlkZW50aXR5Ijoib3BlcmF0b3IiLCJleHAiOjE2NTY4MjIxNDh9.iZ-Sifz1WEkcy1CwX4c-rzI-QgfzUMqpWr2oYr8vZ1o") {
+      id
+      username
+      password
+    }
+  }
+  '''
+
+  r = graph_query(GRAPHQL_URL, query)
+
+  assert r.json()['data']['me']['id'] == '2'
+  assert r.json()['data']['me']['username'] == 'operator'
+  assert r.json()['data']['me']['password'] == '******'