This file is indexed.

/usr/lib/python2.7/dist-packages/ZODB/config.py is in python-zodb 1:3.9.7-5.

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
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Foundation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE
#
##############################################################################
"""Open database and storage from a configuration."""

import os
from cStringIO import StringIO

import ZConfig

import ZODB

db_schema_path = os.path.join(ZODB.__path__[0], "config.xml")
_db_schema = None

s_schema_path = os.path.join(ZODB.__path__[0], "storage.xml")
_s_schema = None

def getDbSchema():
    global _db_schema
    if _db_schema is None:
        _db_schema = ZConfig.loadSchema(db_schema_path)
    return _db_schema

def getStorageSchema():
    global _s_schema
    if _s_schema is None:
        _s_schema = ZConfig.loadSchema(s_schema_path)
    return _s_schema

def databaseFromString(s):
    return databaseFromFile(StringIO(s))

def databaseFromFile(f):
    config, handle = ZConfig.loadConfigFile(getDbSchema(), f)
    return databaseFromConfig(config.database)

def databaseFromURL(url):
    config, handler = ZConfig.loadConfig(getDbSchema(), url)
    return databaseFromConfig(config.database)

def databaseFromConfig(database_factories):
    databases = {}
    first = None
    for factory in database_factories:
        db = factory.open(databases)
        if first is None:
            first = db

    return first

def storageFromString(s):
    return storageFromFile(StringIO(s))

def storageFromFile(f):
    config, handle = ZConfig.loadConfigFile(getStorageSchema(), f)
    return storageFromConfig(config.storage)

def storageFromURL(url):
    config, handler = ZConfig.loadConfig(getStorageSchema(), url)
    return storageFromConfig(config.storage)

def storageFromConfig(section):
    return section.open()

class BaseConfig:
    """Object representing a configured storage or database.

    Methods:

    open() -- open and return the configured object

    Attributes:

    name   -- name of the storage

    """

    def __init__(self, config):
        self.config = config
        self.name = config.getSectionName()

    def open(self, database_name='unnamed', databases=None):
        """Open and return the storage object."""
        raise NotImplementedError

class ZODBDatabase(BaseConfig):

    def open(self, databases=None):
        section = self.config
        storage = section.storage.open()
        options = {}

        def _option(name, oname=None):
            v = getattr(section, name)
            if v is not None:
                if oname is None:
                    oname = name
                options[oname] = v

        _option('pool_timeout')
        _option('allow_implicit_cross_references', 'xrefs')

        try:
            return ZODB.DB(
                storage,
                pool_size=section.pool_size,
                cache_size=section.cache_size,
                cache_size_bytes=section.cache_size_bytes,
                historical_pool_size=section.historical_pool_size,
                historical_cache_size=section.historical_cache_size,
                historical_cache_size_bytes=section.historical_cache_size_bytes,
                historical_timeout=section.historical_timeout,
                database_name=section.database_name or self.name or '',
                databases=databases,
                **options)
        except:
            storage.close()
            raise

class MappingStorage(BaseConfig):

    def open(self):
        from ZODB.MappingStorage import MappingStorage
        return MappingStorage(self.config.name)

class DemoStorage(BaseConfig):

    def open(self):
        base = changes = None
        for factory in self.config.factories:
            if factory.name == 'changes':
                changes = factory.open()
            else:
                if base is None:
                    base = factory.open()
                else:
                    raise ValueError("Too many base storages defined!")

        from ZODB.DemoStorage import DemoStorage
        return DemoStorage(self.config.name, base=base, changes=changes)

class FileStorage(BaseConfig):

    def open(self):
        from ZODB.FileStorage import FileStorage
        config = self.config
        options = {}
        if getattr(config, 'packer', None):
            m, name = config.packer.rsplit('.', 1)
            options['packer'] = getattr(__import__(m, {}, {}, ['*']), name)

        for name in ('blob_dir', 'create', 'read_only', 'quota', 'pack_gc',
                     'pack_keep_old'):
            v = getattr(config, name, self)
            if v is not self:
                options[name] = v

        return FileStorage(config.path, **options)

class BlobStorage(BaseConfig):

    def open(self):
        from ZODB.blob import BlobStorage
        base = self.config.base.open()
        return BlobStorage(self.config.blob_dir, base)


class ZEOClient(BaseConfig):

    def open(self):
        from ZEO.ClientStorage import ClientStorage
        # config.server is a multikey of socket-connection-address values
        # where the value is a socket family, address tuple.
        L = [server.address for server in self.config.server]
        options = {}
        if self.config.blob_cache_size is not None:
            options['blob_cache_size'] = self.config.blob_cache_size
        if self.config.blob_cache_size_check is not None:
            options['blob_cache_size_check'] = self.config.blob_cache_size_check

        return ClientStorage(
            L,
            blob_dir=self.config.blob_dir,
            shared_blob_dir=self.config.shared_blob_dir,
            storage=self.config.storage,
            cache_size=self.config.cache_size,
            name=self.config.name,
            client=self.config.client,
            var=self.config.var,
            min_disconnect_poll=self.config.min_disconnect_poll,
            max_disconnect_poll=self.config.max_disconnect_poll,
            wait=self.config.wait,
            read_only=self.config.read_only,
            read_only_fallback=self.config.read_only_fallback,
            drop_cache_rather_verify=self.config.drop_cache_rather_verify,
            username=self.config.username,
            password=self.config.password,
            realm=self.config.realm,
            **options)

class BDBStorage(BaseConfig):

    def open(self):
        from BDBStorage.BerkeleyBase import BerkeleyConfig
        storageclass = self.get_storageclass()
        bconf = BerkeleyConfig()
        for name in dir(BerkeleyConfig):
            if name.startswith('_'):
                continue
            setattr(bconf, name, getattr(self.config, name))
        return storageclass(self.config.envdir, config=bconf)

class BDBMinimalStorage(BDBStorage):

    def get_storageclass(self):
        import BDBStorage.BDBMinimalStorage
        return BDBStorage.BDBMinimalStorage.BDBMinimalStorage

class BDBFullStorage(BDBStorage):

    def get_storageclass(self):
        import BDBStorage.BDBFullStorage
        return BDBStorage.BDBFullStorage.BDBFullStorage