feat(nginx plugin): add HSTS enhancement (#5463)

* feat(nginx plugin): add HSTS enhancement

* chore(nginx): factor out block-splitting code from redirect & hsts enhancements!

* chore(nginx): merge fixes

* address comments

* fix linter: remove a space

* fix(config): remove SSL directives in HTTP block after block split, and remove_directive removes 'Managed by certbot' comment

* chore(nginx-hsts): Move added SSL directives to a constant on Configurator class

* fix(nginx-hsts): rebase on wildcard cert changes

certbot-nginx/certbot_nginx/configurator.py

@@ -61,6 +61,9 @@ class NginxConfigurator(common.Installer):
 
     DEFAULT_LISTEN_PORT = '80'
 
+    # SSL directives that Certbot can add when installing a new certificate.
+    SSL_DIRECTIVES = ['ssl_certificate', 'ssl_certificate_key', 'ssl_dhparam']
+
     @classmethod
     def add_parser_arguments(cls, add):
         add("server-root", default=constants.CLI_DEFAULTS["server_root"],
@@ -105,6 +108,7 @@ class NginxConfigurator(common.Installer):
         self.parser = None
         self.version = version
         self._enhance_func = {"redirect": self._enable_redirect,
+                              "ensure-http-header": self._set_http_header,
                               "staple-ocsp": self._enable_ocsp_stapling}
 
         self.reverter.recovery_routine()
@@ -621,7 +625,7 @@ class NginxConfigurator(common.Installer):
     ##################################
     def supported_enhancements(self):  # pylint: disable=no-self-use
         """Returns currently supported enhancements."""
-        return ['redirect', 'staple-ocsp']
+        return ['redirect', 'ensure-http-header', 'staple-ocsp']
 
     def enhance(self, domain, enhancement, options=None):
         """Enhance configuration.
@@ -647,6 +651,40 @@ class NginxConfigurator(common.Installer):
         test_redirect_block = _test_block_from_block(_redirect_block_for_domain(domain))
         return vhost.contains_list(test_redirect_block)
 
+    def _set_http_header(self, domain, header_substring):
+        """Enables header identified by header_substring on domain.
+
+        If the vhost is listening plaintextishly, separates out the relevant
+        directives into a new server block, and only add header directive to
+        HTTPS block.
+
+        :param str domain: the domain to enable header for.
+        :param str header_substring: String to uniquely identify a header.
+                        e.g. Strict-Transport-Security, Upgrade-Insecure-Requests
+        :returns: Success
+        :raises .errors.PluginError: If no viable HTTPS host can be created or
+            set with header header_substring.
+        """
+        vhosts = self.choose_vhosts(domain)
+        if not vhosts:
+            raise errors.PluginError(
+                "Unable to find corresponding HTTPS host for enhancement.")
+        for vhost in vhosts:
+            if vhost.has_header(header_substring):
+                raise errors.PluginEnhancementAlreadyPresent(
+                    "Existing %s header" % (header_substring))
+
+            # if there is no separate SSL block, break the block into two and
+            # choose the SSL block.
+            if vhost.ssl and any([not addr.ssl for addr in vhost.addrs]):
+                _, vhost = self._split_block(vhost)
+
+            header_directives = [
+                ['\n    ', 'add_header', ' ', header_substring, ' '] +
+                    constants.HEADER_ARGS[header_substring],
+                ['\n']]
+            self.parser.add_server_directives(vhost, header_directives, replace=False)
+
     def _add_redirect_block(self, vhost, domain):
         """Add redirect directive to vhost
         """
@@ -655,6 +693,39 @@ class NginxConfigurator(common.Installer):
         self.parser.add_server_directives(
             vhost, redirect_block, replace=False, insert_at_top=True)
 
+    def _split_block(self, vhost, only_directives=None):
+        """Splits this "virtual host" (i.e. this nginx server block) into
+        separate HTTP and HTTPS blocks.
+
+        :param vhost: The server block to break up into two.
+        :param list only_directives: If this exists, only duplicate these directives
+            when splitting the block.
+        :type vhost: :class:`~certbot_nginx.obj.VirtualHost`
+        :returns: tuple (http_vhost, https_vhost)
+        :rtype: tuple of type :class:`~certbot_nginx.obj.VirtualHost`
+        """
+        http_vhost = self.parser.duplicate_vhost(vhost, only_directives=only_directives)
+
+        def _ssl_match_func(directive):
+            return 'ssl' in directive
+
+        def _ssl_config_match_func(directive):
+            return self.mod_ssl_conf in directive
+
+        def _no_ssl_match_func(directive):
+            return 'ssl' not in directive
+
+        # remove all ssl addresses and related directives from the new block
+        for directive in self.SSL_DIRECTIVES:
+            self.parser.remove_server_directives(http_vhost, directive)
+        self.parser.remove_server_directives(http_vhost, 'listen', match_func=_ssl_match_func)
+        self.parser.remove_server_directives(http_vhost, 'include',
+                                             match_func=_ssl_config_match_func)
+
+        # remove all non-ssl addresses from the existing block
+        self.parser.remove_server_directives(vhost, 'listen', match_func=_no_ssl_match_func)
+        return http_vhost, vhost
+
     def _enable_redirect(self, domain, unused_options):
         """Redirect all equivalent HTTP traffic to ssl_vhost.
 
@@ -694,28 +765,15 @@ class NginxConfigurator(common.Installer):
         :param `~obj.Vhost` vhost: vhost to enable redirect for
         """
 
-        new_vhost = None
+        http_vhost = None
         if vhost.ssl:
-            new_vhost = self.parser.duplicate_vhost(vhost,
-                only_directives=['listen', 'server_name'])
-
-            def _ssl_match_func(directive):
-                return 'ssl' in directive
-
-            def _no_ssl_match_func(directive):
-                return 'ssl' not in directive
-
-            # remove all ssl addresses from the new block
-            self.parser.remove_server_directives(new_vhost, 'listen', match_func=_ssl_match_func)
-
-            # remove all non-ssl addresses from the existing block
-            self.parser.remove_server_directives(vhost, 'listen', match_func=_no_ssl_match_func)
+            http_vhost, _ = self._split_block(vhost, ['listen', 'server_name'])
 
             # Add this at the bottom to get the right order of directives
             return_404_directive = [['\n    ', 'return', ' ', '404']]
-            self.parser.add_server_directives(new_vhost, return_404_directive, replace=False)
+            self.parser.add_server_directives(http_vhost, return_404_directive, replace=False)
 
-            vhost = new_vhost
+            vhost = http_vhost
 
         if self._has_certbot_redirect(vhost, domain):
             logger.info("Traffic on port %s already redirecting to ssl in %s",

certbot-nginx/certbot_nginx/constants.py

@@ -44,3 +44,7 @@ def os_constant(key):
     :return: value of constant for active os
     """
     return CLI_DEFAULTS[key]
+
+HSTS_ARGS = ['\"max-age=31536000\"', ' ', 'always']
+
+HEADER_ARGS = {'Strict-Transport-Security': HSTS_ARGS}

certbot-nginx/certbot_nginx/obj.py

@@ -5,7 +5,7 @@ import six
 
 from certbot.plugins import common
 
-REDIRECT_DIRECTIVES = ['return', 'rewrite']
+ADD_HEADER_DIRECTIVE = 'add_header'
 
 class Addr(common.Addr):
     r"""Represents an Nginx address, i.e. what comes after the 'listen'
@@ -198,6 +198,14 @@ class VirtualHost(object):  # pylint: disable=too-few-public-methods
                      tuple(self.addrs), tuple(self.names),
                      self.ssl, self.enabled))
 
+    def has_header(self, header_name):
+        """Determine if this server block has a particular header set.
+        :param str header_name: The name of the header to check for, e.g.
+            'Strict-Transport-Security'
+        """
+        found = _find_directive(self.raw, ADD_HEADER_DIRECTIVE, header_name)
+        return found is not None
+
     def contains_list(self, test):
         """Determine if raw server block contains test list at top level
         """
@@ -233,3 +241,19 @@ class VirtualHost(object):  # pylint: disable=too-few-public-methods
                 addrs=", ".join(str(addr) for addr in self.addrs),
                 names=", ".join(self.names),
                 https="Yes" if self.ssl else "No"))
+
+def _find_directive(directives, directive_name, match_content=None):
+    """Find a directive of type directive_name in directives. If match_content is given,
+       Searches for `match_content` in the directive arguments.
+    """
+    if not directives or isinstance(directives, six.string_types) or len(directives) == 0:
+        return None
+
+    # If match_content is None, just match on directive type. Otherwise, match on
+    # both directive type -and- the content!
+    if directives[0] == directive_name and \
+            (match_content is None or match_content in directives):
+        return directives
+
+    matches = (_find_directive(line, directive_name, match_content) for line in directives)
+    return next((m for m in matches if m is not None), None)

certbot-nginx/certbot_nginx/parser.py

@@ -377,6 +377,7 @@ class NginxParser(object):
                     del directive[directive.index('default_server')]
         return new_vhost
 
+
 def _parse_ssl_options(ssl_options):
     if ssl_options is not None:
         try:
@@ -667,6 +668,9 @@ def _add_directive(block, directive, replace, insert_at_top):
     elif block[location] != directive:
         raise errors.MisconfigurationError(err_fmt.format(directive, block[location]))
 
+def _is_certbot_comment(directive):
+    return '#' in directive and COMMENT in directive
+
 def _remove_directives(directive_name, match_func, block):
     """Removes directives of name directive_name from a config block if match_func matches.
     """
@@ -674,6 +678,9 @@ def _remove_directives(directive_name, match_func, block):
         location = _find_location(block, directive_name, match_func=match_func)
         if location is None:
             return
+        # if the directive was made by us, remove the comment following
+        if location + 1 < len(block) and _is_certbot_comment(block[location + 1]):
+            del block[location + 1]
         del block[location]
 
 def _apply_global_addr_ssl(addr_to_ssl, parsed_server):

certbot-nginx/certbot_nginx/tests/configurator_test.py

@@ -94,7 +94,7 @@ class NginxConfiguratorTest(util.NginxTest):
              "globalssl.com", "globalsslsetssl.com", "ipv6.com", "ipv6ssl.com"]))
 
     def test_supported_enhancements(self):
-        self.assertEqual(['redirect', 'staple-ocsp'],
+        self.assertEqual(['redirect', 'ensure-http-header', 'staple-ocsp'],
                          self.config.supported_enhancements())
 
     def test_enhance(self):
@@ -510,6 +510,54 @@ class NginxConfiguratorTest(util.NginxTest):
                ['return', '404'], ['#', ' managed by Certbot'], [], [], []]]],
             generated_conf)
 
+    def test_split_for_headers(self):
+        example_conf = self.config.parser.abs_path('sites-enabled/example.com')
+        self.config.deploy_cert(
+            "example.org",
+            "example/cert.pem",
+            "example/key.pem",
+            "example/chain.pem",
+            "example/fullchain.pem")
+        self.config.enhance("www.example.com", "ensure-http-header", "Strict-Transport-Security")
+        generated_conf = self.config.parser.parsed[example_conf]
+        self.assertEqual(
+            [[['server'], [
+               ['server_name', '.example.com'],
+               ['server_name', 'example.*'], [],
+               ['listen', '5001', 'ssl'], ['#', ' managed by Certbot'],
+               ['ssl_certificate', 'example/fullchain.pem'], ['#', ' managed by Certbot'],
+               ['ssl_certificate_key', 'example/key.pem'], ['#', ' managed by Certbot'],
+               ['include', self.config.mod_ssl_conf], ['#', ' managed by Certbot'],
+               ['ssl_dhparam', self.config.ssl_dhparams], ['#', ' managed by Certbot'],
+               [], [],
+               ['add_header', 'Strict-Transport-Security', '"max-age=31536000"', 'always'],
+               ['#', ' managed by Certbot'],
+               [], []]],
+             [['server'], [
+               ['listen', '69.50.225.155:9000'],
+               ['listen', '127.0.0.1'],
+               ['server_name', '.example.com'],
+               ['server_name', 'example.*'],
+               [], [], []]]],
+            generated_conf)
+
+    def test_http_header_hsts(self):
+        example_conf = self.config.parser.abs_path('sites-enabled/example.com')
+        self.config.enhance("www.example.com", "ensure-http-header",
+                            "Strict-Transport-Security")
+        expected = ['add_header', 'Strict-Transport-Security', '"max-age=31536000"', 'always']
+        generated_conf = self.config.parser.parsed[example_conf]
+        self.assertTrue(util.contains_at_depth(generated_conf, expected, 2))
+
+    def test_http_header_hsts_twice(self):
+        self.config.enhance("www.example.com", "ensure-http-header",
+                            "Strict-Transport-Security")
+        self.assertRaises(
+            errors.PluginEnhancementAlreadyPresent,
+            self.config.enhance, "www.example.com",
+            "ensure-http-header", "Strict-Transport-Security")
+
+
     @mock.patch('certbot_nginx.obj.VirtualHost.contains_list')
     def test_certbot_redirect_exists(self, mock_contains_list):
         # Test that we add no redirect statement if there is already a

certbot-nginx/certbot_nginx/tests/obj_test.py

@@ -143,6 +143,15 @@ class VirtualHostTest(unittest.TestCase):
             "filp",
             set([Addr.fromstring("localhost")]), False, False,
             set(['localhost']), raw4, [])
+        raw_has_hsts = [
+            ['listen', '69.50.225.155:9000'],
+            ['server_name', 'return.com'],
+            ['add_header', 'always', 'set', 'Strict-Transport-Security', '\"max-age=31536000\"'],
+        ]
+        self.vhost_has_hsts = VirtualHost(
+            "filep",
+            set([Addr.fromstring("localhost")]), False, False,
+            set(['localhost']), raw_has_hsts, [])
 
     def test_eq(self):
         from certbot_nginx.obj import Addr
@@ -162,6 +171,12 @@ class VirtualHostTest(unittest.TestCase):
                                  'enabled: False'])
         self.assertEqual(stringified, str(self.vhost1))
 
+    def test_has_header(self):
+        self.assertTrue(self.vhost_has_hsts.has_header('Strict-Transport-Security'))
+        self.assertFalse(self.vhost_has_hsts.has_header('Bogus-Header'))
+        self.assertFalse(self.vhost1.has_header('Strict-Transport-Security'))
+        self.assertFalse(self.vhost1.has_header('Bogus-Header'))
+
     def test_contains_list(self):
         from certbot_nginx.obj import VirtualHost
         from certbot_nginx.obj import Addr

certbot-nginx/certbot_nginx/tests/parser_test.py

@@ -191,6 +191,32 @@ class NginxParserTest(util.NginxTest): #pylint: disable=too-many-public-methods
                           ['server_name', '*.www.foo.com', '*.www.example.com']]
         self.assertTrue(nparser.has_ssl_on_directive(mock_vhost))
 
+
+    def test_remove_server_directives(self):
+        nparser = parser.NginxParser(self.config_path)
+        mock_vhost = obj.VirtualHost(nparser.abs_path('nginx.conf'),
+                                     None, None, None,
+                                     set(['localhost',
+                                           r'~^(www\.)?(example|bar)\.']),
+                                     None, [10, 1, 9])
+        example_com = nparser.abs_path('sites-enabled/example.com')
+        names = set(['.example.com', 'example.*'])
+        mock_vhost.filep = example_com
+        mock_vhost.names = names
+        mock_vhost.path = [0]
+        nparser.add_server_directives(mock_vhost,
+                                      [['foo', 'bar'], ['ssl_certificate',
+                                                        '/etc/ssl/cert2.pem']],
+                                      replace=False)
+        nparser.remove_server_directives(mock_vhost, 'foo')
+        nparser.remove_server_directives(mock_vhost, 'ssl_certificate')
+        self.assertEqual(nparser.parsed[example_com],
+            [[['server'], [['listen', '69.50.225.155:9000'],
+                           ['listen', '127.0.0.1'],
+                           ['server_name', '.example.com'],
+                           ['server_name', 'example.*'],
+                           []]]])
+
     def test_add_server_directives(self):
         nparser = parser.NginxParser(self.config_path)
         mock_vhost = obj.VirtualHost(nparser.abs_path('nginx.conf'),

certbot/constants.py

@@ -135,13 +135,13 @@ RENEWER_DEFAULTS = dict(
 """Defaults for renewer script."""
 
 
-ENHANCEMENTS = ["redirect", "http-header", "ocsp-stapling", "spdy"]
+ENHANCEMENTS = ["redirect", "ensure-http-header", "ocsp-stapling", "spdy"]
 """List of possible :class:`certbot.interfaces.IInstaller`
 enhancements.
 
 List of expected options parameters:
 - redirect: None
-- http-header: TODO
+- ensure-http-header: name of header (i.e. Strict-Transport-Security)
 - ocsp-stapling: certificate chain file path
 - spdy: TODO