This file is indexed.

/usr/lib/python3/dist-packages/oslo_config/cfgfilter.py is in python3-oslo.config 1:3.9.0-3.

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
# Copyright 2014 Red Hat, Inc.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

r"""
There are three use cases for the ConfigFilter class:

1. Help enforce that a given module does not access options registered
   by another module, without first declaring those cross-module
   dependencies using import_opt().

2. Prevent private configuration opts from being visible to modules
   other than the one which registered it.

3. Limit the options on a Cfg object that can be accessed.

.. versionadded:: 1.4

Cross-Module Option Dependencies
--------------------------------

When using the global cfg.CONF object, it is quite common for a module
to require the existence of configuration options registered by other
modules.

For example, if module 'foo' registers the 'blaa' option and the module
'bar' uses the 'blaa' option then 'bar' might do::

  import foo

  print(CONF.blaa)

However, it's completely non-obvious why foo is being imported (is it
unused, can we remove the import) and where the 'blaa' option comes from.

The CONF.import_opt() method allows such a dependency to be explicitly
declared::

  CONF.import_opt('blaa', 'foo')
  print(CONF.blaa)

However, import_opt() has a weakness - if 'bar' imports 'foo' using the
import builtin and doesn't use import_opt() to import 'blaa', then 'blaa'
can still be used without problems. Similarly, where multiple options
are registered a module imported via importopt(), a lazy programmer can
get away with only declaring a dependency on a single option.

The ConfigFilter class provides a way to ensure that options are not
available unless they have been registered in the module or imported using
import_opt() for example with::

  CONF = ConfigFilter(cfg.CONF)
  CONF.import_opt('blaa', 'foo')
  print(CONF.blaa)

no other options other than 'blaa' are available via CONF.

Private Configuration Options
-----------------------------

Libraries which register configuration options typically do not want
users of the library API to access those configuration options. If
API users do access private configuration options, those users will
be disrupted if and when a configuration option is renamed. In other
words, one does not typically wish for the name of the private config
options to be part of the public API.

The ConfigFilter class provides a way for a library to register
options such that they are not visible via the ConfigOpts instance
which the API user supplies to the library. For example::

  from __future__ import print_function

  from oslo_config.cfg import *
  from oslo_config.cfgfilter import *

  class Widget(object):

      def __init__(self, conf):
          self.conf = conf
          self._private_conf = ConfigFilter(self.conf)
          self._private_conf.register_opt(StrOpt('foo'))

      @property
      def foo(self):
          return self._private_conf.foo

  conf = ConfigOpts()
  widget = Widget(conf)
  print(widget.foo)
  print(conf.foo)  # raises NoSuchOptError


Limited Configuration Options
-----------------------------

It may be required that when passing a CONF object to other functions we want
to filter that the receiving code is only able to access a restricted subset
of the options that are available on the CONF object. This is essentially a
more general case of the Private Configuration Options and Cross-Module Options
whereby we expose an option that is already present on the underlying CONF
object without providing any means to load it if not present.

So given a CONF object with options defined::

  CONF.register_opt(StrOpt('foo'))
  CONF.register_opt(StrOpt('bar'))

we can expose options such that only those options are present::

  restricted_conf = CfgFilter(CONF)
  restricted_conf.expose_opt('foo')

  print(restricted_conf.foo)
  print(restricted_conf.bar)  # raises NoSuchOptError
"""

import collections
import itertools

from oslo_config import cfg


class CliOptRegisteredError(cfg.Error):
    """Raised when registering cli opt not in original ConfigOpts.

    .. versionadded:: 1.12
    """

    def __str__(self):
        ret = "Cannot register a cli option that was not present in the" \
              " original ConfigOpts."

        if self.msg:
            ret += ": " + self.msg
        return ret


class ConfigFilter(collections.Mapping):
    """A helper class which wraps a ConfigOpts object.

    ConfigFilter enforces the explicit declaration of dependencies on external
    options and allows private options which are not registered with the
    wrapped Configopts object.
    """

    def __init__(self, conf):
        """Construct a ConfigFilter object.

        :param conf: a ConfigOpts object
        """
        self._conf = conf
        self._fconf = cfg.ConfigOpts()
        self._sync()

        self._imported_opts = set()
        self._imported_groups = dict()

    def _sync(self):
        if self._fconf._namespace is not self._conf._namespace:
            self._fconf.clear()
            self._fconf._namespace = self._conf._namespace
            self._fconf._args = self._conf._args

    def __getattr__(self, name):
        """Look up an option value.

        :param name: the opt name (or 'dest', more precisely)
        :returns: the option value (after string subsititution) or a GroupAttr
        :raises: NoSuchOptError,ConfigFileValueError,TemplateSubstitutionError
        """
        if name in self._imported_groups:
            return self._imported_groups[name]
        elif name in self._imported_opts:
            return getattr(self._conf, name)
        else:
            self._sync()
            return getattr(self._fconf, name)

    def __getitem__(self, key):
        """Look up an option value."""
        return getattr(self, key)

    def __contains__(self, key):
        """Return True if key is the name of a registered opt or group."""
        return (key in self._fconf or
                key in self._imported_opts or
                key in self._imported_groups)

    def __iter__(self):
        """Iterate over all registered opt and group names."""
        return itertools.chain(self._fconf.keys(),
                               self._imported_opts,
                               self._imported_groups.keys())

    def __len__(self):
        """Return the number of options and option groups."""
        return (len(self._fconf) +
                len(self._imported_opts) +
                len(self._imported_groups))

    @staticmethod
    def _already_registered(conf, opt, group=None):
        group_name = group.name if isinstance(group, cfg.OptGroup) else group
        return ((group_name is None and
                 opt.dest in conf) or
                (group_name is not None and
                 group_name in conf and
                 opt.dest in conf[group_name]))

    def register_opt(self, opt, group=None):
        """Register an option schema.

        :param opt: an instance of an Opt sub-class
        :param group: an optional OptGroup object or group name
        :return: False if the opt was already registered, True otherwise
        :raises: DuplicateOptError
        """
        if self._already_registered(self._conf, opt, group):
            # Raises DuplicateError if there is another opt with the same name
            ret = self._conf.register_opt(opt, group)
            self._import_opt(opt.dest, group)
            return ret
        else:
            return self._fconf.register_opt(opt, group)

    def register_opts(self, opts, group=None):
        """Register multiple option schemas at once."""
        for opt in opts:
            self.register_opt(opt, group)

    def register_cli_opt(self, opt, group=None):
        """Register a CLI option schema.

        :param opt: an instance of an Opt sub-class
        :param group: an optional OptGroup object or group name
        :return: False if the opt was already register, True otherwise
        :raises: DuplicateOptError, ArgsAlreadyParsedError
        """
        if self._already_registered(self._conf, opt, group):
            # Raises DuplicateError if there is another opt with the same name
            ret = self._conf.register_opt(
                opt, group, cli=True, clear_cache=False)
            self._import_opt(opt.dest, group)
            return ret
        else:
            raise CliOptRegisteredError(
                "Opt '{0}' cannot be registered.".format(opt.dest))

    def register_cli_opts(self, opts, group=None):
        """Register multiple CLI option schemas at once."""
        for opt in opts:
            self.register_cli_opt(opt, group)

    def register_group(self, group):
        """Register an option group.

        :param group: an OptGroup object
        """
        self._fconf.register_group(group)

    def import_opt(self, opt_name, module_str, group=None):
        """Import an option definition from a module.

        :param name: the name/dest of the opt
        :param module_str: the name of a module to import
        :param group: an option OptGroup object or group name
        :raises: NoSuchOptError, NoSuchGroupError
        """
        self._conf.import_opt(opt_name, module_str, group)
        self._import_opt(opt_name, group)

    def import_group(self, group, module_str):
        """Import an option group from a module.

        Note that this allows access to all options registered with
        the group whether or not those options were registered by
        the given module.

        :param group: an option OptGroup object or group name
        :param module_str: the name of a module to import
        :raises: ImportError, NoSuchGroupError
        """
        self._conf.import_group(group, module_str)
        group = self._import_group(group)
        group._all_opts = True

    def _import_opt(self, opt_name, group):
        if group is None:
            self._imported_opts.add(opt_name)
            return True
        else:
            group = self._import_group(group)
            return group._import_opt(opt_name)

    def _import_group(self, group_or_name):
        if isinstance(group_or_name, cfg.OptGroup):
            group_name = group_or_name.name
        else:
            group_name = group_or_name

        if group_name in self._imported_groups:
            return self._imported_groups[group_name]
        else:
            group = self.GroupAttr(self._conf, group_name)
            self._imported_groups[group_name] = group
            return group

    def expose_opt(self, opt_name, group=None):
        """Expose an option from the underlying conf object.

        This allows an object that has already been imported or used from the
        base conf object to be seen from the filter object.

        :param opt_name: the name/dest of the opt
        :param group: an option OptGroup object or group name
        """
        self._import_opt(opt_name, group)

    def expose_group(self, group):
        """Expose all option from a group in the underlying conf object.

        This allows an object that has already been imported or used from the
        base conf object to be seen from the filter object.

        :param group: an option OptGroup object or group name
        """
        group = self._import_group(group)
        group._all_opts = True

    class GroupAttr(collections.Mapping):

        """Helper class to wrap a group object.

        Represents the option values of a group as a mapping and attributes.
        """

        def __init__(self, conf, group):
            """Construct a GroupAttr object.

            :param conf: a ConfigOpts object
            :param group: an OptGroup object
            """
            self._conf = conf
            self._group = group
            self._imported_opts = set()
            self._all_opts = False

        def __getattr__(self, name):
            """Look up an option value."""
            if not self._all_opts and name not in self._imported_opts:
                raise cfg.NoSuchOptError(name)
            return getattr(self._conf[self._group], name)

        def __getitem__(self, key):
            """Look up an option value."""
            return getattr(self, key)

        def __contains__(self, key):
            """Return True if key is the name of a registered opt or group."""
            return key in self._imported_opts

        def __iter__(self):
            """Iterate over all registered opt and group names."""
            for key in self._imported_opts:
                yield key

        def __len__(self):
            """Return the number of options and option groups."""
            return len(self._imported_opts)

        def _import_opt(self, opt_name):
            self._imported_opts.add(opt_name)