quay/config_app/config_test/test_api_usage.py

from StringIO import StringIO
from mockldap import MockLdap

from data import database, model
from util.security.test.test_ssl_util import generate_test_cert

from config_app.c_app import app
from config_app.config_test import ApiTestCase, all_queues, ADMIN_ACCESS_USER, ADMIN_ACCESS_EMAIL
from config_app.config_endpoints.api import api_bp
from config_app.config_endpoints.api.superuser import SuperUserCustomCertificate, SuperUserCustomCertificates
from config_app.config_endpoints.api.suconfig import SuperUserConfig, SuperUserCreateInitialSuperUser, \
  SuperUserConfigFile, SuperUserRegistryStatus

try:
  app.register_blueprint(api_bp, url_prefix='/api')
except ValueError:
  # This blueprint was already registered
  pass


class TestSuperUserCreateInitialSuperUser(ApiTestCase):
  def test_create_superuser(self):
    data = {
      'username': 'newsuper',
      'password': 'password',
      'email': 'jschorr+fake@devtable.com',
    }

    # Add some fake config.
    fake_config = {
      'AUTHENTICATION_TYPE': 'Database',
      'SECRET_KEY': 'fakekey',
    }

    self.putJsonResponse(SuperUserConfig, data=dict(config=fake_config, hostname='fakehost'))

    # Try to write with config. Should 403 since there are users in the DB.
    self.postResponse(SuperUserCreateInitialSuperUser, data=data, expected_code=403)

    # Delete all users in the DB.
    for user in list(database.User.select()):
      model.user.delete_user(user, all_queues)

    # Create the superuser.
    self.postJsonResponse(SuperUserCreateInitialSuperUser, data=data)

    # Ensure the user exists in the DB.
    self.assertIsNotNone(model.user.get_user('newsuper'))

    # Ensure that the current user is a superuser in the config.
    json = self.getJsonResponse(SuperUserConfig)
    self.assertEquals(['newsuper'], json['config']['SUPER_USERS'])

    # Ensure that the current user is a superuser in memory by trying to call an API
    # that will fail otherwise.
    self.getResponse(SuperUserConfigFile, params=dict(filename='ssl.cert'))


class TestSuperUserConfig(ApiTestCase):
  def test_get_status_update_config(self):
    # With no config the status should be 'config-db'.
    json = self.getJsonResponse(SuperUserRegistryStatus)
    self.assertEquals('config-db', json['status'])

    # Add some fake config.
    fake_config = {
      'AUTHENTICATION_TYPE': 'Database',
      'SECRET_KEY': 'fakekey',
    }

    json = self.putJsonResponse(SuperUserConfig, data=dict(config=fake_config,
                                                           hostname='fakehost'))
    self.assertEquals('fakekey', json['config']['SECRET_KEY'])
    self.assertEquals('fakehost', json['config']['SERVER_HOSTNAME'])
    self.assertEquals('Database', json['config']['AUTHENTICATION_TYPE'])

    # With config the status should be 'setup-db'.
    # TODO: fix this test
    # json = self.getJsonResponse(SuperUserRegistryStatus)
    # self.assertEquals('setup-db', json['status'])

  def test_config_file(self):
    # Try for an invalid file. Should 404.
    self.getResponse(SuperUserConfigFile, params=dict(filename='foobar'), expected_code=404)

    # Try for a valid filename. Should not exist.
    json = self.getJsonResponse(SuperUserConfigFile, params=dict(filename='ssl.cert'))
    self.assertFalse(json['exists'])

    # Add the file.
    self.postResponse(SuperUserConfigFile, params=dict(filename='ssl.cert'),
                      file=(StringIO('my file contents'), 'ssl.cert'))

    # Should now exist.
    json = self.getJsonResponse(SuperUserConfigFile, params=dict(filename='ssl.cert'))
    self.assertTrue(json['exists'])

  def test_update_with_external_auth(self):
    # Run a mock LDAP.
    mockldap = MockLdap({
      'dc=quay,dc=io': {
        'dc': ['quay', 'io']
      },
      'ou=employees,dc=quay,dc=io': {
        'dc': ['quay', 'io'],
        'ou': 'employees'
      },
      'uid=' + ADMIN_ACCESS_USER + ',ou=employees,dc=quay,dc=io': {
        'dc': ['quay', 'io'],
        'ou': 'employees',
        'uid': [ADMIN_ACCESS_USER],
        'userPassword': ['password'],
        'mail': [ADMIN_ACCESS_EMAIL],
      },
    })

    config = {
      'AUTHENTICATION_TYPE': 'LDAP',
      'LDAP_BASE_DN': ['dc=quay', 'dc=io'],
      'LDAP_ADMIN_DN': 'uid=devtable,ou=employees,dc=quay,dc=io',
      'LDAP_ADMIN_PASSWD': 'password',
      'LDAP_USER_RDN': ['ou=employees'],
      'LDAP_UID_ATTR': 'uid',
      'LDAP_EMAIL_ATTR': 'mail',
    }

    mockldap.start()
    try:
      # Write the config with the valid password.
      self.putResponse(SuperUserConfig,
                       data={'config': config,
                             'password': 'password',
                             'hostname': 'foo'}, expected_code=200)

      # Ensure that the user row has been linked.
      # TODO: fix this test
      # self.assertEquals(ADMIN_ACCESS_USER,
      #                   model.user.verify_federated_login('ldap', ADMIN_ACCESS_USER).username)
    finally:
      mockldap.stop()

class TestSuperUserCustomCertificates(ApiTestCase):
  def test_custom_certificates(self):

    # Upload a certificate.
    cert_contents, _ = generate_test_cert(hostname='somecoolhost', san_list=['DNS:bar', 'DNS:baz'])
    self.postResponse(SuperUserCustomCertificate, params=dict(certpath='testcert.crt'),
                      file=(StringIO(cert_contents), 'testcert.crt'), expected_code=204)

    # Make sure it is present.
    json = self.getJsonResponse(SuperUserCustomCertificates)
    self.assertEquals(1, len(json['certs']))

    cert_info = json['certs'][0]
    self.assertEquals('testcert.crt', cert_info['path'])

    self.assertEquals(set(['somecoolhost', 'bar', 'baz']), set(cert_info['names']))
    self.assertFalse(cert_info['expired'])

    # Remove the certificate.
    self.deleteResponse(SuperUserCustomCertificate, params=dict(certpath='testcert.crt'))

    # Make sure it is gone.
    json = self.getJsonResponse(SuperUserCustomCertificates)
    self.assertEquals(0, len(json['certs']))

  def test_expired_custom_certificate(self):
    # Upload a certificate.
    cert_contents, _ = generate_test_cert(hostname='somecoolhost', expires=-10)
    self.postResponse(SuperUserCustomCertificate, params=dict(certpath='testcert.crt'),
                      file=(StringIO(cert_contents), 'testcert.crt'), expected_code=204)

    # Make sure it is present.
    json = self.getJsonResponse(SuperUserCustomCertificates)
    self.assertEquals(1, len(json['certs']))

    cert_info = json['certs'][0]
    self.assertEquals('testcert.crt', cert_info['path'])

    self.assertEquals(set(['somecoolhost']), set(cert_info['names']))
    self.assertTrue(cert_info['expired'])

  def test_invalid_custom_certificate(self):
    # Upload an invalid certificate.
    self.postResponse(SuperUserCustomCertificate, params=dict(certpath='testcert.crt'),
                      file=(StringIO('some contents'), 'testcert.crt'), expected_code=204)

    # Make sure it is present but invalid.
    json = self.getJsonResponse(SuperUserCustomCertificates)
    self.assertEquals(1, len(json['certs']))

    cert_info = json['certs'][0]
    self.assertEquals('testcert.crt', cert_info['path'])
    self.assertEquals('no start line', cert_info['error'])

  def test_path_sanitization(self):
    # Upload a certificate.
    cert_contents, _ = generate_test_cert(hostname='somecoolhost', expires=-10)
    self.postResponse(SuperUserCustomCertificate, params=dict(certpath='testcert/../foobar.crt'),
                      file=(StringIO(cert_contents), 'testcert/../foobar.crt'), expected_code=204)

    # Make sure it is present.
    json = self.getJsonResponse(SuperUserCustomCertificates)
    self.assertEquals(1, len(json['certs']))

    cert_info = json['certs'][0]
    self.assertEquals('foobar.crt', cert_info['path'])