131 lines
4 KiB
Python
131 lines
4 KiB
Python
|
import unittest
|
||
|
|
import base64
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
|
||
|
|
from app import app
|
||
|
|
from data import model
|
||
|
|
from data.database import OAuthApplication, OAuthAccessToken
|
||
|
|
from flask import g
|
||
|
|
from flask.ext.principal import identity_loaded
|
||
|
|
|
||
|
|
from auth.auth import _process_basic_auth
|
||
|
|
from endpoints.api import api_bp, api
|
||
|
|
from endpoints.api.user import User, Signin
|
||
|
|
|
||
|
|
import json as py_json
|
||
|
|
from test.test_api_usage import ApiTestCase
|
||
|
|
|
||
|
|
ADMIN_ACCESS_USER = 'devtable'
|
||
|
|
DISABLED_USER = 'disabled'
|
||
|
|
|
||
|
|
@identity_loaded.connect_via(app)
|
||
|
|
def on_identity_loaded(sender, identity):
|
||
|
|
g.identity = identity
|
||
|
|
|
||
|
|
class TestAuth(ApiTestCase):
|
||
|
|
def verify_cookie_auth(self, username):
|
||
|
|
resp = self.getJsonResponse(User)
|
||
|
|
self.assertEquals(resp['username'], username)
|
||
|
|
|
||
|
|
def verify_identity(self, id):
|
||
|
|
try:
|
||
|
|
identity = g.identity
|
||
|
|
except:
|
||
|
|
identity = None
|
||
|
|
|
||
|
|
self.assertIsNotNone(identity)
|
||
|
|
self.assertEquals(identity.id, id)
|
||
|
|
|
||
|
|
def verify_no_identity(self):
|
||
|
|
try:
|
||
|
|
identity = g.identity
|
||
|
|
except:
|
||
|
|
identity = None
|
||
|
|
|
||
|
|
self.assertIsNone(identity)
|
||
|
|
|
||
|
|
def conduct_basic_auth(self, username, password):
|
||
|
|
encoded = base64.b64encode(username + ':' + password)
|
||
|
|
try:
|
||
|
|
_process_basic_auth('Basic ' + encoded)
|
||
|
|
except:
|
||
|
|
pass
|
||
|
|
|
||
|
|
def create_oauth(self, user):
|
||
|
|
oauth_app = OAuthApplication.create(client_id='onetwothree', redirect_uri='',
|
||
|
|
application_uri='', organization=user,
|
||
|
|
name='someapp')
|
||
|
|
|
||
|
|
expires_at = datetime.utcnow() + timedelta(seconds=50000)
|
||
|
|
OAuthAccessToken.create(application=oauth_app, authorized_user=user,
|
||
|
|
scope='repo:admin',
|
||
|
|
access_token='access1234', token_type='Bearer',
|
||
|
|
expires_at=expires_at, refresh_token=None, data={})
|
||
|
|
|
||
|
|
def test_login(self):
|
||
|
|
password = 'password'
|
||
|
|
resp = self.postJsonResponse(Signin, data=dict(username=ADMIN_ACCESS_USER, password=password))
|
||
|
|
self.assertTrue(resp.get('success'))
|
||
|
|
self.verify_cookie_auth(ADMIN_ACCESS_USER)
|
||
|
|
|
||
|
|
def test_login_disabled(self):
|
||
|
|
password = 'password'
|
||
|
|
self.postJsonResponse(Signin, data=dict(username=DISABLED_USER, password=password),
|
||
|
|
expected_code=403)
|
||
|
|
|
||
|
|
def test_basic_auth_user(self):
|
||
|
|
user = model.get_user(ADMIN_ACCESS_USER)
|
||
|
|
self.conduct_basic_auth(ADMIN_ACCESS_USER, 'password')
|
||
|
|
self.verify_identity(user.uuid)
|
||
|
|
|
||
|
|
def test_basic_auth_disabled_user(self):
|
||
|
|
user = model.get_user(DISABLED_USER)
|
||
|
|
self.conduct_basic_auth(DISABLED_USER, 'password')
|
||
|
|
self.verify_no_identity()
|
||
|
|
|
||
|
|
def test_basic_auth_token(self):
|
||
|
|
token = model.create_delegate_token(ADMIN_ACCESS_USER, 'simple', 'sometoken')
|
||
|
|
self.conduct_basic_auth('$token', token.code)
|
||
|
|
self.verify_identity(token.code)
|
||
|
|
|
||
|
|
def test_basic_auth_invalid_token(self):
|
||
|
|
self.conduct_basic_auth('$token', 'foobar')
|
||
|
|
self.verify_no_identity()
|
||
|
|
|
||
|
|
def test_basic_auth_invalid_user(self):
|
||
|
|
self.conduct_basic_auth('foobarinvalid', 'foobar')
|
||
|
|
self.verify_no_identity()
|
||
|
|
|
||
|
|
def test_oauth_invalid(self):
|
||
|
|
self.conduct_basic_auth('$oauthtoken', 'foobar')
|
||
|
|
self.verify_no_identity()
|
||
|
|
|
||
|
|
def test_oauth_valid_user(self):
|
||
|
|
user = model.get_user(ADMIN_ACCESS_USER)
|
||
|
|
self.create_oauth(user)
|
||
|
|
self.conduct_basic_auth('$oauthtoken', 'access1234')
|
||
|
|
self.verify_identity(user.uuid)
|
||
|
|
|
||
|
|
def test_oauth_disabled_user(self):
|
||
|
|
user = model.get_user(DISABLED_USER)
|
||
|
|
self.create_oauth(user)
|
||
|
|
self.conduct_basic_auth('$oauthtoken', 'access1234')
|
||
|
|
self.verify_no_identity()
|
||
|
|
|
||
|
|
def test_basic_auth_robot(self):
|
||
|
|
user = model.get_user(ADMIN_ACCESS_USER)
|
||
|
|
robot, passcode = model.get_robot('dtrobot', user)
|
||
|
|
self.conduct_basic_auth(robot.username, passcode)
|
||
|
|
self.verify_identity(robot.uuid)
|
||
|
|
|
||
|
|
def test_basic_auth_robot_invalidcode(self):
|
||
|
|
user = model.get_user(ADMIN_ACCESS_USER)
|
||
|
|
robot, _ = model.get_robot('dtrobot', user)
|
||
|
|
self.conduct_basic_auth(robot.username, 'someinvalidcode')
|
||
|
|
self.verify_no_identity()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
unittest.main()
|
||
|
|
|