Coverage for archdocs/features/app_servers/parser.py: 100%

1import re as py_re 

2import types 

3import typing 

4 

5from archdocs import prefilter, settings 

6from archdocs.features.app_servers.const import ApplicationServerEnum, ApplicationServerFeatures, ServerSignature 

7 

8 

9# Servers not found by the module name alone: each has its own entry point, and werkzeug 

10# is also reachable through a literal wider than its package name. 

11_PATTERN_OF_SERVER: typing.Final = types.MappingProxyType( 

12 { 

13 ApplicationServerEnum.tornado_server: r"\btornado\.(?:httpserver|web|ioloop)\b", 

14 ApplicationServerEnum.gevent_server: r"\bgevent\.pywsgi\b", 

15 ApplicationServerEnum.eventlet_server: r"\beventlet\.wsgi\b", 

16 ApplicationServerEnum.werkzeug_server: r"\bwerkzeug\.serving\b|\brun_simple\s*\(", 

17 ApplicationServerEnum.wsgiref_server: r"\bwsgiref\.simple_server\b", 

18 }, 

19) 

20_LITERALS_OF_SERVER: typing.Final = types.MappingProxyType( 

21 {ApplicationServerEnum.werkzeug_server: ("werkzeug", "run_simple")}, 

22) 

23_SERVER_SIGNATURES: typing.Final = tuple( 

24 ServerSignature( 

25 server_name=one_server.value, 

26 prefilter_literals=_LITERALS_OF_SERVER.get(one_server, (one_server.value,)), 

27 compiled_pattern=py_re.compile( 

28 _PATTERN_OF_SERVER.get( 

29 one_server, 

30 rf"\b(?:from|import)\s+{one_server.value}\b|\b{one_server.value}\s+(?:--\w|[\w.]+:\w+)", 

31 ), 

32 flags=settings.TYPICAL_RE_FLAGS, 

33 ), 

34 ) 

35 for one_server in ApplicationServerEnum 

36) 

37_WORKER_CLASS_LITERALS: typing.Final = ("worker_class", "--worker-class") 

38_WORKER_CLASS_PATTERN: typing.Final = py_re.compile( 

39 r"(?:worker_class\s*=|--worker-class)[\s=\"']*(?P<worker_class>[\w.]+)", 

40 flags=settings.TYPICAL_RE_FLAGS, 

41) 

42# The left anchors keep neighbours out of the numbers: `ThreadPoolExecutor(max_workers=10)` 

43# is not ten server workers and `transport = 8080` is not a listen port. 

44_WORKERS_COUNT_PATTERN: typing.Final = py_re.compile( 

45 r"(?:\bworkers\s*=|--workers|\s-w)[\s=\"']*(?P<found_number>\d{1,3})\b", 

46 flags=settings.TYPICAL_RE_FLAGS, 

47) 

48_LISTEN_PORT_PATTERN: typing.Final = py_re.compile( 

49 r"(?:(?:(?<![a-z])port\s*=|--port)[\s=\"']*|\bbind[\s=\"']*[\w.\[\]]*:)(?P<found_number>\d{2,5})\b", 

50 flags=settings.TYPICAL_RE_FLAGS, 

51) 

52_TLS_PATTERN: typing.Final = py_re.compile( 

53 r"\bssl_cert\w*\b|\bssl_key\w*\b|\bcertfile\b|\bkeyfile\b|\bssl_context\b|--ssl-\w+", 

54 flags=settings.TYPICAL_RE_FLAGS, 

55) 

56_HTTP2_PATTERN: typing.Final = py_re.compile( 

57 r"\bhttp2\b|\bhttp\s*=\s*[\"']?2\b|--http[\s=]+2\b", 

58 flags=settings.TYPICAL_RE_FLAGS, 

59) 

60 

61 

62def _read_first_number(number_pattern: py_re.Pattern[str], raw_source: str, /) -> int: 

63 number_match: typing.Final = number_pattern.search(raw_source) 

64 if number_match is None: 

65 return 0 

66 return int(number_match.group("found_number")) 

67 

68 

69def _find_servers_behind_worker_class(raw_source: str, lowered_source: str, /) -> set[str]: 

70 if not prefilter.contains_any_literal(lowered_source, _WORKER_CLASS_LITERALS): 

71 return set() 

72 worker_class_match: typing.Final = _WORKER_CLASS_PATTERN.search(raw_source) 

73 if worker_class_match is None: 

74 return set() 

75 worker_class_name: typing.Final = worker_class_match.group("worker_class").lower() 

76 return {ApplicationServerEnum.gunicorn_server.value} | { 

77 one_server.value for one_server in ApplicationServerEnum if one_server.value in worker_class_name 

78 } 

79 

80 

81def find_application_server_features(raw_source: str) -> ApplicationServerFeatures: 

82 lowered_source: typing.Final = raw_source.lower() 

83 servers_found: typing.Final = { 

84 one_signature.server_name 

85 for one_signature in _SERVER_SIGNATURES 

86 if prefilter.contains_any_literal(lowered_source, one_signature.prefilter_literals) 

87 and one_signature.compiled_pattern.search(raw_source) 

88 } | _find_servers_behind_worker_class(raw_source, lowered_source) 

89 if not servers_found: 

90 return ApplicationServerFeatures() 

91 return ApplicationServerFeatures( 

92 servers_used=frozenset(servers_found), 

93 workers_count=_read_first_number(_WORKERS_COUNT_PATTERN, raw_source), 

94 listen_port=_read_first_number(_LISTEN_PORT_PATTERN, raw_source), 

95 tls_enabled=bool(_TLS_PATTERN.search(raw_source)), 

96 http2_enabled=bool(_HTTP2_PATTERN.search(raw_source)), 

97 )