| 49 | |
| 50 | |
| 51 | class VersionRequirement: |
| 52 | WHITESPACE_REGEX = re.compile(r"\s+") |
| 53 | SSM_SESSION_PLUGIN_VERSION_REGEX = re.compile(r"^\d+(\.\d+){0,3}$") |
| 54 | |
| 55 | def __init__(self, min_version): |
| 56 | self.min_version = min_version |
| 57 | |
| 58 | def meets_requirement(self, version): |
| 59 | ssm_plugin_version = self._sanitize_plugin_version(version) |
| 60 | if self._is_valid_version(ssm_plugin_version): |
| 61 | norm_version, norm_min_version = self._normalize( |
| 62 | ssm_plugin_version, self.min_version |
| 63 | ) |
| 64 | return norm_version > norm_min_version |
| 65 | else: |
| 66 | return False |
| 67 | |
| 68 | def _sanitize_plugin_version(self, plugin_version): |
| 69 | return re.sub(self.WHITESPACE_REGEX, "", plugin_version) |
| 70 | |
| 71 | def _is_valid_version(self, plugin_version): |
| 72 | return bool( |
| 73 | self.SSM_SESSION_PLUGIN_VERSION_REGEX.match(plugin_version) |
| 74 | ) |
| 75 | |
| 76 | def _normalize(self, v1, v2): |
| 77 | v1_parts = [int(v) for v in v1.split(".")] |
| 78 | v2_parts = [int(v) for v in v2.split(".")] |
| 79 | while len(v1_parts) != len(v2_parts): |
| 80 | if len(v1_parts) - len(v2_parts) > 0: |
| 81 | v2_parts.append(0) |
| 82 | else: |
| 83 | v1_parts.append(0) |
| 84 | return v1_parts, v2_parts |
| 85 | |
| 86 | |
| 87 | class StartSessionCommand(ServiceOperation): |