quay/util/config/validators/validate_secscan.py

import time

from app import app
from boot import setup_jwt_proxy
from util.secscan.api import SecurityScannerAPI
from util.config.validators import BaseValidator, ConfigValidationException

class SecurityScannerValidator(BaseValidator):
  name = "security-scanner"

  @classmethod
  def validate(cls, config, user, user_password):
    """ Validates the configuration for talking to a Quay Security Scanner. """
    if not config.get('FEATURE_SECURITY_SCANNER', False):
      return

    client = app.config['HTTPCLIENT']
    api = SecurityScannerAPI(app, config, None, client=client, skip_validation=True)

    if not config.get('TESTING', False):
      # Generate a temporary Quay key to use for signing the outgoing requests.
      setup_jwt_proxy()

    # We have to wait for JWT proxy to restart with the newly generated key.
    max_tries = 5
    response = None
    last_exception = None

    while max_tries > 0:
      try:
        response = api.ping()
        last_exception = None
        if response.status_code == 200:
          return
      except Exception as ex:
        last_exception = ex

      time.sleep(1)
      max_tries = max_tries - 1

    if last_exception is not None:
      message = str(last_exception)
      raise ConfigValidationException('Could not ping security scanner: %s' % message)
    else:
      message = 'Expected 200 status code, got %s: %s' % (response.status_code, response.text)
      raise ConfigValidationException('Could not ping security scanner: %s' % message)