test_webserver/test/test_helper.rb

require "minitest/autorun"
require "minitest/reporters"
Minitest::Reporters.use!


module SSLHelper
  def check_ssl_cert_cmd(domain:, rootcert: nil, openssl: openssl_path(:system), issuer: nil, warning: 45, critical: 15, cn: nil, ocsp: true, host_cn: true)
    # check_ssl_cert is a Nagios plugin, usable outside of Nagios
    # cf. https://trac.id.ethz.ch/projects/nagios_plugins/wiki/check_ssl_cert

    args = []
    args.push("--host", domain)
    args.push("--rootcert", rootcert) if rootcert
    args.push("--openssl", openssl) if openssl
    args.push("--issuer", issuer) if issuer
    args.push("--warning", warning) if warning
    args.push("--critical", critical) if critical
    args.push("--cn", cn) if cn
    args.push("--ocsp") if ocsp
    args.push("--host-cn") if host_cn

    "vendor/check_ssl_cert/check_ssl_cert #{args.join(" ")}"
  end

  def analyze_cmd(domain:, level: "intermediate", openssl: openssl_path(:system))
    # Cipherscan helps audit SSL configuration
    # cf. https://github.com/jvehent/cipherscan

    args = [
      "-o", openssl,
      "-l", level,
      "-t", domain
    ]

    "vendor/cipherscan/analyze.py #{args.join(' ')}"
  end

  def openssl_verify_cmd(domain, options = "")
    args = [
      "-CAfile", "#{root_certificate}",
      "-connect", "#{domain}:443",
      options,
      "2>&1",
    ]

    "echo QUIT | #{openssl_path} s_client #{args.join(' ')}"
  end

  def openssl_path(variant = :system)
    case variant
    when :local
      "vendor/cipherscan/openssl-darwin64"
    else
      `which openssl`.chop
    end
  end

  def root_certificate
    "test/certs/AddTrust_External_CA_Root.pem"
  end
end

module WebserverHelper

  def message_with_context(message, context = nil)
    if context.nil? || context.empty?
      message
    else
      message + " #{context}"
    end
  end

  def webserver_env
    webserver_env = ENV['WEBSERVER_ENV']
    webserver_env = "production" if webserver_env.nil? || webserver_env.empty?
    if %w(production staging development).include?(webserver_env)
      webserver_env
    else
      fail ArgumentError, "Environnement #{webserver_env} invalide"
    end
  end

  def domain
    case webserver_env
    when "production"
      "www.example.com"
    when "staging"
      "www-staging.example.com"
    when "development"
      "local.example.com"
    else
      fail ArgumentError, "Domaine indéterminé"
    end
  end

  def base_url
    "https://#{domain}"
  end

  def internal_url?(url)
    URI(url).host[/example.com\Z/]
  end

  def agent
    @agent ||= Mechanize.new
  end

  def page_and_doc(url)
    page = agent.get(url)
    doc = Nokogiri::HTML(page.body)

    if block_given?
      yield(page, doc)
    else
      [page, doc]
    end
  end

  def on_page(path = "/", &block)
    uri = URI.join(base_url, path)
    page, doc = page_and_doc(uri)

    yield(page, doc)
  end

  def on_home_page(&block)
    on_page("/", &block)
  end

  def on_pages(pages = [], &block)
    pages.each do |page|
      on_page(page, &block)
    end
  end

  # Custom assertions

  def assert_scheme(scheme, url, context = nil)
    uri = URI.parse(url)
    assert_equal scheme, uri.scheme, message_with_context("Expected scheme to be '#{scheme}' for '#{url}'", context)
  end

  def assert_cachable_asset(page, context = nil)
    assert_status_ok page, context
    assert_max_age "315360000", page, context
    assert_public page, context
    assert_has_etag page, context
    assert_has_last_modified page, context
  end

  # def assert_hsts(page, context = nil)
  #   assert_status_ok page, context
  #   assert_hsts_max_age "315360000", page, context
  #   assert_public page, context
  #   refute_hsts_include_subdomains page, context
  # end

  def assert_has_header(header, page, context = nil)
    assert page.response.key?(header), message_with_context("Expected to find '#{header}' header".freeze, context)
  end

  def assert_has_hsts(page, context = nil)
    assert_includes page.response.keys, "strict-transport-security", message_with_context("Expected to find an Strict-Transport-Security header", context)
  end

  def assert_hsts_max_age(expected, page, context = nil)
    assert_equal expected, hsts_max_age(page), message_with_context("Expected Strict-Transport-Security 'max-age' directive to be #{expected}", context)
  end

  def assert_hsts_include_subdomains(page, context = nil)
    refute hsts_include_subdomains?(page), message_with_context("Expected not to find Strict-Transport-Security 'includeSubdomains' directive", context)
  end

  def assert_has_etag(page, context = nil)
    assert_includes page.response.keys, "etag", message_with_context("Expected to find an ETag header", context)
  end
  def refute_has_etag(page, context = nil)
    refute_includes page.response.keys, "etag", message_with_context("Expected not to find an ETag header", context)
  end

  def assert_has_last_modified(page, context = nil)
    assert_includes page.response.keys, "last-modified", message_with_context("Expected to find a Last-Modified header", context)
  end

  def assert_max_age(expected, page, context = nil)
    assert_equal expected, cache_max_age(page), message_with_context("Expected Cache-Control 'max-age' directive to be #{expected}", context)
  end

  def assert_must_revalidate(page, context = nil)
    assert cache_must_revalidate?(page), message_with_context("Expected Cache-Control 'must-revalidate' directive to be found", context)
  end

  def refute_must_revalidate(page, context = nil)
    refute cache_must_revalidate?(page), message_with_context("Expected Cache-Control 'must-revalidate' directive not to be found", context)
  end

  def assert_private(page, context = nil)
    assert cache_private?(page), message_with_context("Expected Cache-Control directive to be private", context)
  end

  def assert_public(page, context = nil)
    assert cache_public?(page), message_with_context("Expected Cache-Control directive to be public", context)
  end

  def assert_status_ok(page, context = nil)
    assert_code("200", page, context)
  end

  def assert_status_not_modified(page, context = nil)
    assert_code("304", page, context)
  end

  def assert_code(expected_code, page_or_code, context = nil)
    actual_code = page_or_code.respond_to?(:code) ? page_or_code.code : page_or_code
    assert_equal expected_code, actual_code, message_with_context("Expected HTTP status code to be #{expected_code}", context)
  end

  def assert_has_x_cache(page, context = nil)
    assert_includes page.response.keys, "x-cache", message_with_context("Expected to find an X-Cache header", context)
  end
  def assert_x_cache_hit(page, context = nil)
    assert_equal "HIT", x_cache_header(page), message_with_context("Expected X-Cache header to be be HIT", context)
  end
  def assert_x_cache_miss(page, context = nil)
    assert_equal "MISS", x_cache_header(page), message_with_context("Expected X-Cache header to be be MISS", context)
  end

  # Helper methods

  def header(page, key)
    page.response[key]
  end

  def status_header(page)
    header(page, "status".freeze)
  end

  def hsts_header(page)
    header(page, "strict-transport-security".freeze)
  end

  def hsts_directives(page)
    hsts_header(page).downcase.split(',').map(&:strip)
  end

  def hsts_include_subdomains?(page)
    cache_control_directives(page).include?("includeSubdomains".freeze)
  end

  def hsts_max_age(page)
    pattern = /\Amax-age\s?=\s?(\d+)\Z/
    if found = hsts_directives(page).detect("") { |v| pattern =~ v }
      found[pattern, 1]
    end
  end

  def etag_header(page)
    header(page, "etag".freeze)
  end

  def last_modified_header(page)
    header(page, "last-modified".freeze)
  end

  def cache_control_header(page)
    header(page, "cache-control".freeze)
  end

  def x_cache_header(page)
    header(page, "x-cache".freeze)
  end

  def cache_private?(page)
    cache_control_directives(page).include?("private")
  end

  def cache_public?(page)
    cache_control_directives(page).include?("public")
  end

  def cache_no_cache?(page)
    cache_control_header(page).downcase.strip == "no-cache"
  end

  def cache_must_revalidate?(page)
    cache_control_directives(page).include?("must-revalidate")
  end

  def cache_has_max_age?(page)
    cache_control_directives(page).any? { |v| v[/\Amax-age=\d\Z/] }
  end

  def cache_control_directives(page)
    cache_control_header(page).downcase.split(',').map(&:strip)
  end

  def cache_max_age(page)
    pattern = /\Amax-age\s?=\s?(\d+)\Z/
    if found = cache_control_directives(page).detect("") { |v| pattern =~ v }
      found[pattern, 1]
    end
  end

end