/usr/lib/python3/dist-packages/certbot_nginx/parser.py is in python3-certbot-nginx 0.28.0-1~deb9u1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 | """NginxParser is a member object of the NginxConfigurator class."""
import copy
import functools
import glob
import logging
import os
import pyparsing
import re
import six
from certbot import errors
from certbot_nginx import obj
from certbot_nginx import nginxparser
from acme.magic_typing import Union, Dict, Set, Any, List, Tuple # pylint: disable=unused-import, no-name-in-module
logger = logging.getLogger(__name__)
class NginxParser(object):
"""Class handles the fine details of parsing the Nginx Configuration.
:ivar str root: Normalized absolute path to the server root
directory. Without trailing slash.
:ivar dict parsed: Mapping of file paths to parsed trees
"""
def __init__(self, root):
self.parsed = {} # type: Dict[str, Union[List, nginxparser.UnspacedList]]
self.root = os.path.abspath(root)
self.config_root = self._find_config_root()
# Parse nginx.conf and included files.
# TODO: Check sites-available/ as well. For now, the configurator does
# not enable sites from there.
self.load()
def load(self):
"""Loads Nginx files into a parsed tree.
"""
self.parsed = {}
self._parse_recursively(self.config_root)
def _parse_recursively(self, filepath):
"""Parses nginx config files recursively by looking at 'include'
directives inside 'http' and 'server' blocks. Note that this only
reads Nginx files that potentially declare a virtual host.
:param str filepath: The path to the files to parse, as a glob
"""
filepath = self.abs_path(filepath)
trees = self._parse_files(filepath)
for tree in trees:
for entry in tree:
if _is_include_directive(entry):
# Parse the top-level included file
self._parse_recursively(entry[1])
elif entry[0] == ['http'] or entry[0] == ['server']:
# Look for includes in the top-level 'http'/'server' context
for subentry in entry[1]:
if _is_include_directive(subentry):
self._parse_recursively(subentry[1])
elif entry[0] == ['http'] and subentry[0] == ['server']:
# Look for includes in a 'server' context within
# an 'http' context
for server_entry in subentry[1]:
if _is_include_directive(server_entry):
self._parse_recursively(server_entry[1])
def abs_path(self, path):
"""Converts a relative path to an absolute path relative to the root.
Does nothing for paths that are already absolute.
:param str path: The path
:returns: The absolute path
:rtype: str
"""
if not os.path.isabs(path):
return os.path.join(self.root, path)
else:
return path
def _build_addr_to_ssl(self):
"""Builds a map from address to whether it listens on ssl in any server block
"""
servers = self._get_raw_servers()
addr_to_ssl = {} # type: Dict[Tuple[str, str], bool]
for filename in servers:
for server, _ in servers[filename]:
# Parse the server block to save addr info
parsed_server = _parse_server_raw(server)
for addr in parsed_server['addrs']:
addr_tuple = addr.normalized_tuple()
if addr_tuple not in addr_to_ssl:
addr_to_ssl[addr_tuple] = addr.ssl
addr_to_ssl[addr_tuple] = addr.ssl or addr_to_ssl[addr_tuple]
return addr_to_ssl
def _get_raw_servers(self):
# pylint: disable=cell-var-from-loop
# type: () -> Dict
"""Get a map of unparsed all server blocks
"""
servers = {} # type: Dict[str, Union[List, nginxparser.UnspacedList]]
for filename in self.parsed:
tree = self.parsed[filename]
servers[filename] = []
srv = servers[filename] # workaround undefined loop var in lambdas
# Find all the server blocks
_do_for_subarray(tree, lambda x: len(x) >= 2 and x[0] == ['server'],
lambda x, y: srv.append((x[1], y)))
# Find 'include' statements in server blocks and append their trees
for i, (server, path) in enumerate(servers[filename]):
new_server = self._get_included_directives(server)
servers[filename][i] = (new_server, path)
return servers
def get_vhosts(self):
# pylint: disable=cell-var-from-loop
"""Gets list of all 'virtual hosts' found in Nginx configuration.
Technically this is a misnomer because Nginx does not have virtual
hosts, it has 'server blocks'.
:returns: List of :class:`~certbot_nginx.obj.VirtualHost`
objects found in configuration
:rtype: list
"""
enabled = True # We only look at enabled vhosts for now
servers = self._get_raw_servers()
vhosts = []
for filename in servers:
for server, path in servers[filename]:
# Parse the server block into a VirtualHost object
parsed_server = _parse_server_raw(server)
vhost = obj.VirtualHost(filename,
parsed_server['addrs'],
parsed_server['ssl'],
enabled,
parsed_server['names'],
server,
path)
vhosts.append(vhost)
self._update_vhosts_addrs_ssl(vhosts)
return vhosts
def _update_vhosts_addrs_ssl(self, vhosts):
"""Update a list of raw parsed vhosts to include global address sslishness
"""
addr_to_ssl = self._build_addr_to_ssl()
for vhost in vhosts:
for addr in vhost.addrs:
addr.ssl = addr_to_ssl[addr.normalized_tuple()]
if addr.ssl:
vhost.ssl = True
def _get_included_directives(self, block):
"""Returns array with the "include" directives expanded out by
concatenating the contents of the included file to the block.
:param list block:
:rtype: list
"""
result = copy.deepcopy(block) # Copy the list to keep self.parsed idempotent
for directive in block:
if _is_include_directive(directive):
included_files = glob.glob(
self.abs_path(directive[1]))
for incl in included_files:
try:
result.extend(self.parsed[incl])
except KeyError:
pass
return result
def _parse_files(self, filepath, override=False):
"""Parse files from a glob
:param str filepath: Nginx config file path
:param bool override: Whether to parse a file that has been parsed
:returns: list of parsed tree structures
:rtype: list
"""
files = glob.glob(filepath) # nginx on unix calls glob(3) for this
# XXX Windows nginx uses FindFirstFile, and
# should have a narrower call here
trees = []
for item in files:
if item in self.parsed and not override:
continue
try:
with open(item) as _file:
parsed = nginxparser.load(_file)
self.parsed[item] = parsed
trees.append(parsed)
except IOError:
logger.warning("Could not open file: %s", item)
except pyparsing.ParseException as err:
logger.debug("Could not parse file: %s due to %s", item, err)
return trees
def _find_config_root(self):
"""Return the Nginx Configuration Root file."""
location = ['nginx.conf']
for name in location:
if os.path.isfile(os.path.join(self.root, name)):
return os.path.join(self.root, name)
raise errors.NoInstallationError(
"Could not find Nginx root configuration file (nginx.conf)")
def filedump(self, ext='tmp', lazy=True):
"""Dumps parsed configurations into files.
:param str ext: The file extension to use for the dumped files. If
empty, this overrides the existing conf files.
:param bool lazy: Only write files that have been modified
"""
# Best-effort atomicity is enforced above us by reverter.py
for filename in self.parsed:
tree = self.parsed[filename]
if ext:
filename = filename + os.path.extsep + ext
try:
if lazy and not tree.is_dirty():
continue
out = nginxparser.dumps(tree)
logger.debug('Writing nginx conf tree to %s:\n%s', filename, out)
with open(filename, 'w') as _file:
_file.write(out)
except IOError:
logger.error("Could not open file for writing: %s", filename)
def parse_server(self, server):
"""Parses a list of server directives, accounting for global address sslishness.
:param list server: list of directives in a server block
:rtype: dict
"""
addr_to_ssl = self._build_addr_to_ssl()
parsed_server = _parse_server_raw(server)
_apply_global_addr_ssl(addr_to_ssl, parsed_server)
return parsed_server
def has_ssl_on_directive(self, vhost):
"""Does vhost have ssl on for all ports?
:param :class:`~certbot_nginx.obj.VirtualHost` vhost: The vhost in question
:returns: True if 'ssl on' directive is included
:rtype: bool
"""
server = vhost.raw
for directive in server:
if not directive:
continue
elif _is_ssl_on_directive(directive):
return True
return False
def add_server_directives(self, vhost, directives, insert_at_top=False):
"""Add directives to the server block identified by vhost.
This method modifies vhost to be fully consistent with the new directives.
..note :: It's an error to try and add a nonrepeatable directive that already
exists in the config block with a conflicting value.
..todo :: Doesn't match server blocks whose server_name directives are
split across multiple conf files.
:param :class:`~certbot_nginx.obj.VirtualHost` vhost: The vhost
whose information we use to match on
:param list directives: The directives to add
:param bool insert_at_top: True if the directives need to be inserted at the top
of the server block instead of the bottom
"""
self._modify_server_directives(vhost,
functools.partial(_add_directives, directives, insert_at_top))
def update_or_add_server_directives(self, vhost, directives, insert_at_top=False):
"""Add or replace directives in the server block identified by vhost.
This method modifies vhost to be fully consistent with the new directives.
..note :: When a directive with the same name already exists in the
config block, the first instance will be replaced. Otherwise, the directive
will be appended/prepended to the config block as in add_server_directives.
..todo :: Doesn't match server blocks whose server_name directives are
split across multiple conf files.
:param :class:`~certbot_nginx.obj.VirtualHost` vhost: The vhost
whose information we use to match on
:param list directives: The directives to add
:param bool insert_at_top: True if the directives need to be inserted at the top
of the server block instead of the bottom
"""
self._modify_server_directives(vhost,
functools.partial(_update_or_add_directives, directives, insert_at_top))
def remove_server_directives(self, vhost, directive_name, match_func=None):
"""Remove all directives of type directive_name.
:param :class:`~certbot_nginx.obj.VirtualHost` vhost: The vhost
to remove directives from
:param string directive_name: The directive type to remove
:param callable match_func: Function of the directive that returns true for directives
to be deleted.
"""
self._modify_server_directives(vhost,
functools.partial(_remove_directives, directive_name, match_func))
def _update_vhost_based_on_new_directives(self, vhost, directives_list):
new_server = self._get_included_directives(directives_list)
parsed_server = self.parse_server(new_server)
vhost.addrs = parsed_server['addrs']
vhost.ssl = parsed_server['ssl']
vhost.names = parsed_server['names']
vhost.raw = new_server
def _modify_server_directives(self, vhost, block_func):
filename = vhost.filep
try:
result = self.parsed[filename]
for index in vhost.path:
result = result[index]
if not isinstance(result, list) or len(result) != 2:
raise errors.MisconfigurationError("Not a server block.")
result = result[1]
block_func(result)
self._update_vhost_based_on_new_directives(vhost, result)
except errors.MisconfigurationError as err:
raise errors.MisconfigurationError("Problem in %s: %s" % (filename, str(err)))
def duplicate_vhost(self, vhost_template, remove_singleton_listen_params=False,
only_directives=None):
"""Duplicate the vhost in the configuration files.
:param :class:`~certbot_nginx.obj.VirtualHost` vhost_template: The vhost
whose information we copy
:param bool remove_singleton_listen_params: If we should remove parameters
from listen directives in the block that can only be used once per address
:param list only_directives: If it exists, only duplicate the named directives. Only
looks at first level of depth; does not expand includes.
:returns: A vhost object for the newly created vhost
:rtype: :class:`~certbot_nginx.obj.VirtualHost`
"""
# TODO: https://github.com/certbot/certbot/issues/5185
# put it in the same file as the template, at the same level
new_vhost = copy.deepcopy(vhost_template)
enclosing_block = self.parsed[vhost_template.filep]
for index in vhost_template.path[:-1]:
enclosing_block = enclosing_block[index]
raw_in_parsed = copy.deepcopy(enclosing_block[vhost_template.path[-1]])
if only_directives is not None:
new_directives = nginxparser.UnspacedList([])
for directive in raw_in_parsed[1]:
if len(directive) > 0 and directive[0] in only_directives:
new_directives.append(directive)
raw_in_parsed[1] = new_directives
self._update_vhost_based_on_new_directives(new_vhost, new_directives)
enclosing_block.append(raw_in_parsed)
new_vhost.path[-1] = len(enclosing_block) - 1
if remove_singleton_listen_params:
for addr in new_vhost.addrs:
addr.default = False
addr.ipv6only = False
for directive in enclosing_block[new_vhost.path[-1]][1]:
if len(directive) > 0 and directive[0] == 'listen':
# Exclude one-time use parameters which will cause an error if repeated.
# https://nginx.org/en/docs/http/ngx_http_core_module.html#listen
exclude = set(('default_server', 'default', 'setfib', 'fastopen', 'backlog',
'rcvbuf', 'sndbuf', 'accept_filter', 'deferred', 'bind',
'ipv6only', 'reuseport', 'so_keepalive'))
for param in exclude:
# See: github.com/certbot/certbot/pull/6223#pullrequestreview-143019225
keys = [x.split('=')[0] for x in directive]
if param in keys:
del directive[keys.index(param)]
return new_vhost
def _parse_ssl_options(ssl_options):
if ssl_options is not None:
try:
with open(ssl_options) as _file:
return nginxparser.load(_file)
except IOError:
logger.warn("Missing NGINX TLS options file: %s", ssl_options)
except pyparsing.ParseBaseException as err:
logger.debug("Could not parse file: %s due to %s", ssl_options, err)
return []
def _do_for_subarray(entry, condition, func, path=None):
"""Executes a function for a subarray of a nested array if it matches
the given condition.
:param list entry: The list to iterate over
:param function condition: Returns true iff func should be executed on item
:param function func: The function to call for each matching item
"""
if path is None:
path = []
if isinstance(entry, list):
if condition(entry):
func(entry, path)
else:
for index, item in enumerate(entry):
_do_for_subarray(item, condition, func, path + [index])
def get_best_match(target_name, names):
"""Finds the best match for target_name out of names using the Nginx
name-matching rules (exact > longest wildcard starting with * >
longest wildcard ending with * > regex).
:param str target_name: The name to match
:param set names: The candidate server names
:returns: Tuple of (type of match, the name that matched)
:rtype: tuple
"""
exact = []
wildcard_start = []
wildcard_end = []
regex = []
for name in names:
if _exact_match(target_name, name):
exact.append(name)
elif _wildcard_match(target_name, name, True):
wildcard_start.append(name)
elif _wildcard_match(target_name, name, False):
wildcard_end.append(name)
elif _regex_match(target_name, name):
regex.append(name)
if len(exact) > 0:
# There can be more than one exact match; e.g. eff.org, .eff.org
match = min(exact, key=len)
return ('exact', match)
if len(wildcard_start) > 0:
# Return the longest wildcard
match = max(wildcard_start, key=len)
return ('wildcard_start', match)
if len(wildcard_end) > 0:
# Return the longest wildcard
match = max(wildcard_end, key=len)
return ('wildcard_end', match)
if len(regex) > 0:
# Just return the first one for now
match = regex[0]
return ('regex', match)
return (None, None)
def _exact_match(target_name, name):
return target_name == name or '.' + target_name == name
def _wildcard_match(target_name, name, start):
# Degenerate case
if name == '*':
return True
parts = target_name.split('.')
match_parts = name.split('.')
# If the domain ends in a wildcard, do the match procedure in reverse
if not start:
parts.reverse()
match_parts.reverse()
# The first part must be a wildcard or blank, e.g. '.eff.org'
first = match_parts.pop(0)
if first != '*' and first != '':
return False
target_name = '.'.join(parts)
name = '.'.join(match_parts)
# Ex: www.eff.org matches *.eff.org, eff.org does not match *.eff.org
return target_name.endswith('.' + name)
def _regex_match(target_name, name):
# Must start with a tilde
if len(name) < 2 or name[0] != '~':
return False
# After tilde is a perl-compatible regex
try:
regex = re.compile(name[1:])
if re.match(regex, target_name):
return True
else:
return False
except re.error: # pragma: no cover
# perl-compatible regexes are sometimes not recognized by python
return False
def _is_include_directive(entry):
"""Checks if an nginx parsed entry is an 'include' directive.
:param list entry: the parsed entry
:returns: Whether it's an 'include' directive
:rtype: bool
"""
return (isinstance(entry, list) and
len(entry) == 2 and entry[0] == 'include' and
isinstance(entry[1], six.string_types))
def _is_ssl_on_directive(entry):
"""Checks if an nginx parsed entry is an 'ssl on' directive.
:param list entry: the parsed entry
:returns: Whether it's an 'ssl on' directive
:rtype: bool
"""
return (isinstance(entry, list) and
len(entry) == 2 and entry[0] == 'ssl' and
entry[1] == 'on')
def _add_directives(directives, insert_at_top, block):
"""Adds directives to a config block."""
for directive in directives:
_add_directive(block, directive, insert_at_top)
if block and '\n' not in block[-1]: # could be " \n " or ["\n"] !
block.append(nginxparser.UnspacedList('\n'))
def _update_or_add_directives(directives, insert_at_top, block):
"""Adds or replaces directives in a config block."""
for directive in directives:
_update_or_add_directive(block, directive, insert_at_top)
if block and '\n' not in block[-1]: # could be " \n " or ["\n"] !
block.append(nginxparser.UnspacedList('\n'))
INCLUDE = 'include'
REPEATABLE_DIRECTIVES = set(['server_name', 'listen', INCLUDE, 'rewrite', 'add_header'])
COMMENT = ' managed by Certbot'
COMMENT_BLOCK = [' ', '#', COMMENT]
def comment_directive(block, location):
"""Add a ``#managed by Certbot`` comment to the end of the line at location.
:param list block: The block containing the directive to be commented
:param int location: The location within ``block`` of the directive to be commented
"""
next_entry = block[location + 1] if location + 1 < len(block) else None
if isinstance(next_entry, list) and next_entry:
if len(next_entry) >= 2 and next_entry[-2] == "#" and COMMENT in next_entry[-1]:
return
elif isinstance(next_entry, nginxparser.UnspacedList):
next_entry = next_entry.spaced[0]
else:
next_entry = next_entry[0]
block.insert(location + 1, COMMENT_BLOCK[:])
if next_entry is not None and "\n" not in next_entry:
block.insert(location + 2, '\n')
def _comment_out_directive(block, location, include_location):
"""Comment out the line at location, with a note of explanation."""
comment_message = ' duplicated in {0}'.format(include_location)
# add the end comment
# create a dumpable object out of block[location] (so it includes the ;)
directive = block[location]
new_dir_block = nginxparser.UnspacedList([]) # just a wrapper
new_dir_block.append(directive)
dumped = nginxparser.dumps(new_dir_block)
commented = dumped + ' #' + comment_message # add the comment directly to the one-line string
new_dir = nginxparser.loads(commented) # reload into UnspacedList
# add the beginning comment
insert_location = 0
if new_dir[0].spaced[0] != new_dir[0][0]: # if there's whitespace at the beginning
insert_location = 1
new_dir[0].spaced.insert(insert_location, "# ") # comment out the line
new_dir[0].spaced.append(";") # directly add in the ;, because now dumping won't work properly
dumped = nginxparser.dumps(new_dir)
new_dir = nginxparser.loads(dumped) # reload into an UnspacedList
block[location] = new_dir[0] # set the now-single-line-comment directive back in place
def _find_location(block, directive_name, match_func=None):
"""Finds the index of the first instance of directive_name in block.
If no line exists, use None."""
return next((index for index, line in enumerate(block) \
if line and line[0] == directive_name and (match_func is None or match_func(line))), None)
def _is_whitespace_or_comment(directive):
"""Is this directive either a whitespace or comment directive?"""
return len(directive) == 0 or directive[0] == '#'
def _add_directive(block, directive, insert_at_top):
if not isinstance(directive, nginxparser.UnspacedList):
directive = nginxparser.UnspacedList(directive)
if _is_whitespace_or_comment(directive):
# whitespace or comment
block.append(directive)
return
location = _find_location(block, directive[0])
# Append or prepend directive. Fail if the name is not a repeatable directive name,
# and there is already a copy of that directive with a different value
# in the config file.
# handle flat include files
directive_name = directive[0]
def can_append(loc, dir_name):
""" Can we append this directive to the block? """
return loc is None or (isinstance(dir_name, six.string_types)
and dir_name in REPEATABLE_DIRECTIVES)
err_fmt = 'tried to insert directive "{0}" but found conflicting "{1}".'
# Give a better error message about the specific directive than Nginx's "fail to restart"
if directive_name == INCLUDE:
# in theory, we might want to do this recursively, but in practice, that's really not
# necessary because we know what file we're talking about (and if we don't recurse, we
# just give a worse error message)
included_directives = _parse_ssl_options(directive[1])
for included_directive in included_directives:
included_dir_loc = _find_location(block, included_directive[0])
included_dir_name = included_directive[0]
if not _is_whitespace_or_comment(included_directive) \
and not can_append(included_dir_loc, included_dir_name):
if block[included_dir_loc] != included_directive:
raise errors.MisconfigurationError(err_fmt.format(included_directive,
block[included_dir_loc]))
else:
_comment_out_directive(block, included_dir_loc, directive[1])
if can_append(location, directive_name):
if insert_at_top:
# Add a newline so the comment doesn't comment
# out existing directives
block.insert(0, nginxparser.UnspacedList('\n'))
block.insert(0, directive)
comment_directive(block, 0)
else:
block.append(directive)
comment_directive(block, len(block) - 1)
elif block[location] != directive:
raise errors.MisconfigurationError(err_fmt.format(directive, block[location]))
def _update_directive(block, directive, location):
block[location] = directive
comment_directive(block, location)
def _update_or_add_directive(block, directive, insert_at_top):
if not isinstance(directive, nginxparser.UnspacedList):
directive = nginxparser.UnspacedList(directive)
if _is_whitespace_or_comment(directive):
# whitespace or comment
block.append(directive)
return
location = _find_location(block, directive[0])
# we can update directive
if location is not None:
_update_directive(block, directive, location)
return
_add_directive(block, directive, insert_at_top)
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.
"""
while True:
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):
"""Apply global sslishness information to the parsed server block
"""
for addr in parsed_server['addrs']:
addr.ssl = addr_to_ssl[addr.normalized_tuple()]
if addr.ssl:
parsed_server['ssl'] = True
def _parse_server_raw(server):
"""Parses a list of server directives.
:param list server: list of directives in a server block
:rtype: dict
"""
addrs = set() # type: Set[obj.Addr]
ssl = False # type: bool
names = set() # type: Set[str]
apply_ssl_to_all_addrs = False
for directive in server:
if not directive:
continue
if directive[0] == 'listen':
addr = obj.Addr.fromstring(" ".join(directive[1:]))
if addr:
addrs.add(addr)
if addr.ssl:
ssl = True
elif directive[0] == 'server_name':
names.update(x.strip('"\'') for x in directive[1:])
elif _is_ssl_on_directive(directive):
ssl = True
apply_ssl_to_all_addrs = True
if apply_ssl_to_all_addrs:
for addr in addrs:
addr.ssl = True
return {
'addrs': addrs,
'ssl': ssl,
'names': names
}
|