forked from QubesOS/qubes-mgmt-salt-base-topd
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmatcher.py
More file actions
281 lines (218 loc) · 6.47 KB
/
Copy pathmatcher.py
File metadata and controls
281 lines (218 loc) · 6.47 KB
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
# -*- coding: utf-8 -*-
#
# vim: set ts=4 sw=4 sts=4 et :
'''
:maintainer: Jason Mehring <nrgaway@gmail.com>
:maturity: new
:depends: none
:platform: all
'''
from __future__ import absolute_import
# Import python libs
import collections.abc
import functools
import logging
import operator
import re
from itertools import (chain, compress, ) # pylint: disable=E0598
# Import salt libs
from salt.utils.datastructures import OrderedDict
# Enable logging
log = logging.getLogger(__name__)
try:
__context__['salt.loaded.ext.util.matcher'] = True
except NameError:
__context__ = {}
# Define the module's virtual name
__virtualname__ = 'matcher'
DEFAULT_PATTERN = [r'.*']
def __virtual__():
'''
'''
return __virtualname__
class Regex(str):
'''
Wrapper to be able to identify regex expressions
'''
pass
def getter(index, element, *ignored): # pylint: disable=W0613
if isinstance(element, collections.abc.Mapping):
getter_ = operator.itemgetter
else:
getter_ = operator.attrgetter
if not index:
return lambda x: ()
if isinstance(index, list):
return getter_(*index)
else:
return getter_(index)
def extract_labels(element=None, *ignored): # pylint: disable=W0613
'''
Return an element's labels.
Uses dictionary keys for a dictionary, _fields for a namedtuple and
index number for list or regular tuple.
Args:
element:
'''
if not element:
return []
try:
# OrderedDict
return element.keys()
except AttributeError:
try:
# namedtuple
return element._fields
except AttributeError:
pass
return element
def generate_selectors(labels=None, *fields, **kwargs):
'''
Create an element list based in another objects labels that will create
a value of True in the corresponding element if in either selectors
or kwargs, otherwise False.
Args:
labels:
Example:
>>> labels = ['one', 'two', 'three', 'four']
>>> fields = ['two', 'three']
>>> generate_selectors(labels, fields)
[False, True, True, False]
'''
if not labels:
return []
enabled = True if 'all' in fields or 'all' in kwargs else False
selectors = [enabled for i in range(len(labels))] # pylint: disable=W0612
if enabled:
return selectors
for index, selector in enumerate(labels):
if selector in fields or selector in kwargs:
selectors[index] = True
return selectors
def translate(pattern):
'''
Translate a shell PATTERN to a regular expression. Based on
'fnmatch.translate'.
Args:
pattern:
Note:
There is no way to quote meta-characters.
'''
i, n = 0, len(pattern)
res = ''
while i < n:
c = pattern[i]
i += 1
if c == '*':
res += '.*'
elif c == '?':
res += '.'
elif c == '[':
j = i
if j < n and pattern[j] == '!':
j += 1
if j < n and pattern[j] == ']':
j += 1
while j < n and pattern[j] != ']':
j += 1
if j >= n:
res += '\\['
else:
stuff = pattern[i:j].replace('\\', '\\\\')
i = j + 1
if stuff[0] == '!':
stuff = '^' + stuff[1:]
elif stuff[0] == '^':
stuff = '\\' + stuff
res = '{0}[{1}]'.format(res, stuff)
else:
res = res + re.escape(c)
return res
def escape_text(text, regex=False):
'''
Escape text for regex pattern match.
Args:
text:
regex:
'''
# Don't escape regex strings as they are assumed to be proper syntax
if isinstance(text, Regex):
return text
elif regex:
return re.escape(text)
return translate(text)
def get_default_pattern(regex): # pylint: disable=W0613
return DEFAULT_PATTERN
def compile(labels, **patterns): # pylint: disable=W0622
'''
Compile patterns.
Args:
labels:
'''
pattern = patterns.pop('_pattern', None)
if pattern:
return pattern
regex = patterns.pop('_regex', False)
escape = patterns.pop('_escape', [])
if not patterns or not labels:
return None
for pattern in list(patterns.keys()):
if pattern not in labels:
patterns.pop(pattern)
default_pattern = get_default_pattern(regex)
escape = escape if escape else []
_escape_text = functools.partial(escape_text, regex=regex)
# Set default values and join patterns for each field
pattern = OrderedDict.fromkeys(labels, None)
for label in labels:
if label in patterns and patterns[label]:
field = patterns[label]
if isinstance(field, re.Pattern): # pylint: disable=W0212
field = [field.pattern]
if isinstance(field, str):
field = [field]
if label in escape or not regex:
field = [_escape_text(text) for text in field]
else:
field = default_pattern
pattern[label] = r'(?:{0})'.format(r'|'.join(field))
try:
return re.compile(
r'\n'.join(pattern.values()), re.MULTILINE | re.DOTALL
)
except NameError:
raise
def itext(element):
'''
Converts element to a text string suitable for regex parsing.
Args:
element:
'''
# Dictionary
if isinstance(element, collections.abc.Mapping):
return '\n'.join(map(str, element.values()))
# Tuple / list
else:
return '\n'.join(map(str, element))
def match(sequence, pattern):
'''
Regex match.
Args:
sequence:
Either a string, list of strings or list of lists / tuples
pattern:
'''
if not pattern:
return chain(sequence)
# Match to text string created from element
return map(pattern.match, map(itext, sequence))
def get_pattern(sequence=None, *ignored, **patterns): # pylint: disable=W0613
if '_pattern' in patterns:
return patterns['_pattern']
labels = extract_labels(sequence)
return compile(labels, **patterns)
def ifilter(sequence, **patterns):
pattern = get_pattern(*sequence, **patterns)
return compress(sequence, match(sequence, pattern))
def filter(sequence, **patterns): # pylint: disable=W0622
return list(ifilter(sequence, **patterns))