|
19 | 19 | .. versionchanged:: 1.3.5 make API public to use as interface |
20 | 20 | """ |
21 | 21 |
|
22 | | -__all__ = ("ExecHelper", "ExecuteAsyncResult") |
| 22 | +__all__ = ("ExecHelper", "ExecuteAsyncResult", "mask_command") |
23 | 23 |
|
24 | 24 | # Standard Library |
25 | 25 | import abc |
@@ -86,6 +86,28 @@ def __exit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any |
86 | 86 | self._conn.__exit__(exc_type=exc_type, exc_val=exc_val, exc_tb=exc_tb) # type: ignore |
87 | 87 |
|
88 | 88 |
|
| 89 | +def mask_command(text: str, rules: str) -> str: |
| 90 | + """Mask part of text using rules. |
| 91 | +
|
| 92 | + :param text: source text |
| 93 | + :param rules: regex rules to mask. |
| 94 | + :return: source with all MATCHED groups replaced by '<*masked*>' |
| 95 | + """ |
| 96 | + masked: typing.List[str] = [] |
| 97 | + |
| 98 | + # places to exclude |
| 99 | + prev = 0 |
| 100 | + for match in re.finditer(rules, text): |
| 101 | + for idx, _ in enumerate(match.groups(), start=1): |
| 102 | + start, end = match.span(idx) |
| 103 | + masked.append(text[prev:start]) |
| 104 | + masked.append("<*masked*>") |
| 105 | + prev = end |
| 106 | + masked.append(text[prev:]) |
| 107 | + |
| 108 | + return "".join(masked) |
| 109 | + |
| 110 | + |
89 | 111 | class ExecHelper(metaclass=abc.ABCMeta): |
90 | 112 | """ExecHelper global API.""" |
91 | 113 |
|
@@ -191,38 +213,12 @@ def _mask_command(self, cmd: str, log_mask_re: typing.Optional[str] = None) -> s |
191 | 213 | .. versionadded:: 1.2.0 |
192 | 214 | """ |
193 | 215 |
|
194 | | - def mask(text: str, rules: str) -> str: |
195 | | - """Mask part of text using rules. |
196 | | -
|
197 | | - :param text: source text |
198 | | - :param rules: regex rules to mask. |
199 | | - :return: source with all MATCHED groups replaced by '<*masked*>' |
200 | | - """ |
201 | | - indexes: typing.List[int] = [0] # Start of the line |
202 | | - masked: typing.List[str] = [] |
203 | | - |
204 | | - # places to exclude |
205 | | - for match in re.finditer(rules, text): |
206 | | - for idx, _ in enumerate(match.groups()): |
207 | | - indexes.extend(match.span(idx + 1)) |
208 | | - indexes.append(len(text)) # End |
209 | | - |
210 | | - # Replace inserts |
211 | | - for idx in range(0, len(indexes) - 2, 2): |
212 | | - start: int = indexes[idx] |
213 | | - end: int = indexes[idx + 1] |
214 | | - masked.append(text[start:end] + "<*masked*>") |
215 | | - |
216 | | - # noinspection PyPep8 |
217 | | - masked.append(text[indexes[-2] : indexes[-1]]) # final part |
218 | | - return "".join(masked) |
219 | | - |
220 | 216 | result: str = cmd.rstrip() |
221 | 217 |
|
222 | 218 | if self.log_mask_re is not None: |
223 | | - result = mask(result, self.log_mask_re) |
| 219 | + result = mask_command(result, self.log_mask_re) |
224 | 220 | if log_mask_re is not None: |
225 | | - result = mask(result, log_mask_re) |
| 221 | + result = mask_command(result, log_mask_re) |
226 | 222 |
|
227 | 223 | return result |
228 | 224 |
|
|
0 commit comments