Add support to ExternalJWT Auth for external user linking

test/test_external_jwt_authn.py

@@ -53,6 +53,58 @@ class JWTAuthTestCase(LiveServerTestCase):
       data = base64.b64decode(request.headers['Authorization'][len('Basic '):])
       return data.split(':', 1)
 
+    @jwt_app.route('/user/query', methods=['GET'])
+    def query_users():
+      query = request.args.get('query')
+      results = []
+
+      for user in users:
+        if user['name'].startswith(query):
+          results.append({
+            'username': user['name'],
+            'email': user['email'],
+          })
+
+      token_data = {
+        'iss': 'authy',
+        'aud': 'quay.io/jwtauthn/query',
+        'nbf': datetime.utcnow(),
+        'iat': datetime.utcnow(),
+        'exp': datetime.utcnow() + timedelta(seconds=60),
+        'results': results,
+      }
+
+      encoded = jwt.encode(token_data, private_key, 'RS256')
+      return jsonify({
+        'token': encoded
+      })
+
+    @jwt_app.route('/user/get', methods=['GET'])
+    def get_user():
+      username = request.args.get('username')
+
+      if username == 'disabled':
+        return make_response('User is currently disabled', 401)
+
+      for user in users:
+        if user['name'] == username or user['email'] == username:
+          token_data = {
+            'iss': 'authy',
+            'aud': 'quay.io/jwtauthn/getuser',
+            'nbf': datetime.utcnow(),
+            'iat': datetime.utcnow(),
+            'exp': datetime.utcnow() + timedelta(seconds=60),
+            'sub': user['name'],
+            'email': user['email']
+          }
+
+          encoded = jwt.encode(token_data, private_key, 'RS256')
+          return jsonify({
+            'token': encoded
+          })
+
+      return make_response('Invalid username or password', 404)
+
     @jwt_app.route('/user/verify', methods=['GET'])
     def verify_user():
       username, password = _get_basic_auth()
@@ -80,7 +132,7 @@ class JWTAuthTestCase(LiveServerTestCase):
             'token': encoded
           })
 
-      return make_response('', 404)
+      return make_response('Invalid username or password', 404)
 
     jwt_app.config['TESTING'] = True
     return jwt_app
@@ -94,7 +146,11 @@ class JWTAuthTestCase(LiveServerTestCase):
 
     self.session = requests.Session()
 
-    self.jwt_auth = ExternalJWTAuthN(self.get_server_url() + '/user/verify', 'authy', '',
+    verify_url = self.get_server_url() + '/user/verify'
+    query_url = self.get_server_url() + '/user/query'
+    getuser_url = self.get_server_url() + '/user/get'
+
+    self.jwt_auth = ExternalJWTAuthN(verify_url, query_url, getuser_url, 'authy', '',
                                      app.config['HTTPCLIENT'], 300, JWTAuthTestCase.public_key.name)
 
   def tearDown(self):
@@ -142,11 +198,78 @@ class JWTAuthTestCase(LiveServerTestCase):
     self.assertIsNotNone(result)
     self.assertEquals('some_neat_user', result.username)
 
-  def test_disabled_user_custom_erorr(self):
+  def test_disabled_user_custom_error(self):
     result, error_message = self.jwt_auth.verify_and_link_user('disabled', 'password')
     self.assertIsNone(result)
     self.assertEquals('User is currently disabled', error_message)
 
+  def test_query(self):
+    # Lookup `cool`.
+    results, identifier, error_message = self.jwt_auth.query_users('cool')
+    self.assertIsNone(error_message)
+    self.assertEquals('jwtauthn', identifier)
+    self.assertEquals(1, len(results))
+
+    self.assertEquals('cooluser', results[0].username)
+    self.assertEquals('user@domain.com', results[0].email)
+
+    # Lookup `some`.
+    results, identifier, error_message = self.jwt_auth.query_users('some')
+    self.assertIsNone(error_message)
+    self.assertEquals('jwtauthn', identifier)
+    self.assertEquals(1, len(results))
+
+    self.assertEquals('some.neat.user', results[0].username)
+    self.assertEquals('neat@domain.com', results[0].email)
+
+    # Lookup `unknown`.
+    results, identifier, error_message = self.jwt_auth.query_users('unknown')
+    self.assertIsNone(error_message)
+    self.assertEquals('jwtauthn', identifier)
+    self.assertEquals(0, len(results))
+
+  def test_get_user(self):
+    # Lookup cooluser.
+    result, error_message = self.jwt_auth.get_user('cooluser')
+    self.assertIsNone(error_message)
+    self.assertIsNotNone(result)
+
+    self.assertEquals('cooluser', result.username)
+    self.assertEquals('user@domain.com', result.email)
+
+    # Lookup some.neat.user.
+    result, error_message = self.jwt_auth.get_user('some.neat.user')
+    self.assertIsNone(error_message)
+    self.assertIsNotNone(result)
+
+    self.assertEquals('some.neat.user', result.username)
+    self.assertEquals('neat@domain.com', result.email)
+
+    # Lookup unknown user.
+    result, error_message = self.jwt_auth.get_user('unknownuser')
+    self.assertIsNone(result)
+
+  def test_link_user(self):
+    # Link cooluser.
+    user, error_message = self.jwt_auth.link_user('cooluser')
+    self.assertIsNone(error_message)
+    self.assertIsNotNone(user)
+    self.assertEquals('cooluser', user.username)
+
+    # Link again. Should return the same user record.
+    user_again, _ = self.jwt_auth.link_user('cooluser')
+    self.assertEquals(user_again.id, user.id)
+
+    # Confirm cooluser.
+    result, _ = self.jwt_auth.confirm_existing_user('cooluser', 'password')
+    self.assertIsNotNone(result)
+    self.assertEquals('cooluser', result.username)
+
+  def test_link_invalid_user(self):
+    user, error_message = self.jwt_auth.link_user('invaliduser')
+    self.assertIsNotNone(error_message)
+    self.assertIsNone(user)
+
 
 if __name__ == '__main__':
   unittest.main()