Skip to content

Z
Zope
Commit 8e149c4b authored by Hanno Schlichting's avatar Hanno Schlichting
Browse files
Options

Move ZServer related starter logic into ZServer project.

parent efe61189
Hide whitespace changes
Inline Side-by-side
Showing with 160 additions and 1217 deletions
src/Zope2/Startup/__init__.py
View file @ 8e149c4b
......@@ -11,433 +11,30 @@
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
""" Startup package. Responsible for startup configuration of Zope """
import logging
import os
import sys
import socket
from re import compile
from socket import gethostbyaddr
import ZConfig
from ZConfig.components.logger import loghandler
from zope.event import notify
from zope.processlifetime import ProcessStarting
from __future__ import absolute_import
import Zope2.Startup.config
try:
IO_ERRORS = (IOError, WindowsError)
except NameError:
IO_ERRORS = (IOError,)
import sys
from zope.deferredimport import deprecated
logger = logging.getLogger("Zope")
started = False
# BBB Zope 5.0
deprecated(
'Please import from ZServer.Startup.starter',
UnixZopeStarter='ZServer.Zope2.Startup.starter:UnixZopeStarter',
WindowsZopeStarter='ZServer.Zope2.Startup.starter:WindowsZopeStarter',
ZopeStarter='ZServer.Zope2.Startup.starter:ZopeStarter',
)
def get_starter(wsgi=False):
if sys.platform[:3].lower() == "win":
if wsgi:
return WindowsWSGIStarter()
else:
return WindowsZopeStarter()
if wsgi:
from Zope2.Startup.starter import WSGIStarter
return WSGIStarter()
else:
if wsgi:
return UnixWSGIStarter()
if sys.platform[:3].lower() == "win":
from ZServer.Zope2.Startup.starter import WindowsZopeStarter
return WindowsZopeStarter()
else:
from ZServer.Zope2.Startup.starter import UnixZopeStarter
return UnixZopeStarter()
class WSGIStarter(object):
"""This is a class which starts Zope as a WSGI app."""
wsgi = True
def __init__(self):
self.event_logger = logging.getLogger()
# We log events to the root logger, which is backed by a
# "StartupHandler" log handler. The "StartupHandler" buffers
# log messages. When the "real" loggers are set up, we flush
# accumulated messages in StartupHandler's buffers to the real
# logger.
formatter = logging.Formatter(
"%(asctime)s %(levelname)s %(name)s %(message)s",
"%Y-%m-%d %H:%M:%S")
self.debug_handler = loghandler.StreamHandler()
self.debug_handler.setFormatter(formatter)
self.debug_handler.setLevel(logging.WARN)
self.startup_handler = loghandler.StartupHandler()
self.event_logger.addHandler(self.debug_handler)
self.event_logger.addHandler(self.startup_handler)
def prepare(self):
self.setupInitialLogging()
self.setupLocale()
self.setupSecurityOptions()
self.setupPublisher()
self.setupFinalLogging()
self.setupInterpreter()
self.startZope()
from App.config import getConfiguration
config = getConfiguration() # NOQA
self.registerSignals()
logger.info('Ready to handle requests')
self.sendEvents()
def getLoggingLevel(self):
if self.cfg.eventlog is None:
level = logging.INFO
else:
# get the lowest handler level. This is the effective level
# level at which which we will spew messages to the console
# during startup.
level = self.cfg.eventlog.getLowestHandlerLevel()
return level
def registerSignals(self):
from Signals import Signals
Signals.registerZopeSignals([self.cfg.eventlog,
self.cfg.access,
self.cfg.trace])
def sendEvents(self):
notify(ProcessStarting())
def setConfiguration(self, cfg):
self.cfg = cfg
def setupConfiguredLoggers(self):
# Must happen after ZopeStarter.setupInitialLogging()
self.event_logger.removeHandler(self.startup_handler)
if self.cfg.eventlog is not None:
self.cfg.eventlog()
if self.cfg.access is not None:
self.cfg.access()
if self.cfg.trace is not None:
self.cfg.trace()
# flush buffered startup messages to event logger
if self.cfg.debug_mode:
self.event_logger.removeHandler(self.debug_handler)
self.startup_handler.flushBufferTo(self.event_logger)
self.event_logger.addHandler(self.debug_handler)
else:
self.startup_handler.flushBufferTo(self.event_logger)
def setupFinalLogging(self):
pass
def setupInitialLogging(self):
if self.cfg.debug_mode:
self.debug_handler.setLevel(self.getLoggingLevel())
else:
self.event_logger.removeHandler(self.debug_handler)
self.debug_handler = None
def setupInterpreter(self):
# make changes to the python interpreter environment
sys.setcheckinterval(self.cfg.python_check_interval)
def setupLocale(self):
# set a locale if one has been specified in the config
if not self.cfg.locale:
return
# workaround to allow unicode encoding conversions in DTML
import codecs
dummy = codecs.lookup('utf-8') # NOQA
locale_id = self.cfg.locale
if locale_id is not None:
try:
import locale
except:
raise ZConfig.ConfigurationError(
'The locale module could not be imported.\n'
'To use localization options, you must ensure\n'
'that the locale module is compiled into your\n'
'Python installation.')
try:
locale.setlocale(locale.LC_ALL, locale_id)
except:
raise ZConfig.ConfigurationError(
'The specified locale "%s" is not supported by your'
'system.\nSee your operating system documentation for '
'more\ninformation on locale support.' % locale_id)
def setupPublisher(self):
import ZPublisher.HTTPRequest
import ZPublisher.Publish
ZPublisher.Publish.set_default_debug_mode(self.cfg.debug_mode)
ZPublisher.Publish.set_default_authentication_realm(
self.cfg.http_realm)
if self.cfg.trusted_proxies:
mapped = []
for name in self.cfg.trusted_proxies:
mapped.extend(_name2Ips(name))
ZPublisher.HTTPRequest.trusted_proxies = tuple(mapped)
Zope2.Startup.config.TRUSTED_PROXIES = tuple(mapped)
def setupSecurityOptions(self):
import AccessControl
AccessControl.setImplementation(
self.cfg.security_policy_implementation)
AccessControl.setDefaultBehaviors(
not self.cfg.skip_ownership_checking,
not self.cfg.skip_authentication_checking,
self.cfg.verbose_security)
def startZope(self):
# Import Zope
import Zope2
Zope2.startup()
class WindowsWSGIStarter(WSGIStarter):
def setupInitialLogging(self):
super(WindowsWSGIStarter, self).setupInitialLogging()
self.setupConfiguredLoggers()
class UnixWSGIStarter(WSGIStarter):
def setupInitialLogging(self):
super(UnixWSGIStarter, self).setupInitialLogging()
level = self.getLoggingLevel()
self.startup_handler.setLevel(level)
# set the initial logging level (this will be changed by the
# ZConfig settings later)
self.event_logger.setLevel(level)
def setupFinalLogging(self):
self.setupConfiguredLoggers()
class ZopeStarter(WSGIStarter):
"""This is a class which starts Zope with a ZServer."""
wsgi = False
def prepare(self):
self.setupInitialLogging()
self.setupLocale()
self.setupSecurityOptions()
self.setupPublisher()
# Start ZServer servers before we drop privileges so we can bind
# to "low" ports:
self.setupZServer()
self.setupServers()
# drop privileges after setting up servers
self.dropPrivileges()
self.setupFinalLogging()
self.makeLockFile()
self.makePidFile()
self.setupInterpreter()
self.startZope()
self.serverListen()
from App.config import getConfiguration
config = getConfiguration() # NOQA
self.registerSignals()
logger.info('Ready to handle requests')
self.sendEvents()
def dropPrivileges(self):
return dropPrivileges(self.cfg)
def run(self):
# the mainloop.
try:
from App.config import getConfiguration
config = getConfiguration() # NOQA
import Lifetime
Lifetime.loop()
from Zope2.Startup.config import ZSERVER_EXIT_CODE
sys.exit(ZSERVER_EXIT_CODE)
finally:
self.shutdown()
def shutdown(self):
databases = getattr(self.cfg.dbtab, 'databases', {})
for db in databases.values():
db.close()
self.unlinkLockFile()
self.unlinkPidFile()
# XXX does anyone actually use these three?
def info(self, msg):
logger.info(msg)
def panic(self, msg):
logger.critical(msg)
def error(self, msg):
logger.error(msg)
def setupZServer(self):
# Increase the number of threads
Zope2.Startup.config.setNumberOfThreads(self.cfg.zserver_threads)
Zope2.Startup.config.ZSERVER_CONNECTION_LIMIT = \
self.cfg.max_listen_sockets
def serverListen(self):
for server in self.cfg.servers:
if hasattr(server, 'fast_listen'):
# This one has the delayed listening feature
if not server.fast_listen:
server.fast_listen = True
# same value as defined in medusa.http_server.py
server.listen(1024)
def setupServers(self):
socket_err = (
'There was a problem starting a server of type "%s". '
'This may mean that your user does not have permission to '
'bind to the port which the server is trying to use or the '
'port may already be in use by another application. '
'(%s)')
servers = []
for server in self.cfg.servers:
# create the server from the server factory
# set up in the config
try:
servers.append(server.create())
except socket.error as e:
raise ZConfig.ConfigurationError(
socket_err % (server.servertype(), e[1]))
self.cfg.servers = servers
def makeLockFile(self):
if not self.cfg.zserver_read_only_mode:
# lock_file is used for the benefit of zctl-like systems, so they
# can tell whether Zope is already running before attempting to
# fire it off again.
#
# We aren't concerned about locking the file to protect against
# other Zope instances running from our CLIENT_HOME, we just
# try to lock the file to signal that zctl should not try to
# start Zope if *it* can't lock the file; we don't panic
# if we can't lock it.
# we need a separate lock file because on win32, locks are not
# advisory, otherwise we would just use the pid file
from Zope2.Startup.misc.lock_file import lock_file
lock_filename = self.cfg.lock_filename
try:
if os.path.exists(lock_filename):
os.unlink(lock_filename)
self.lockfile = open(lock_filename, 'w')
lock_file(self.lockfile)
self.lockfile.write(str(os.getpid()))
self.lockfile.flush()
except IO_ERRORS:
pass
def makePidFile(self):
if not self.cfg.zserver_read_only_mode:
# write the pid into the pidfile if possible
try:
if os.path.exists(self.cfg.pid_filename):
os.unlink(self.cfg.pid_filename)
f = open(self.cfg.pid_filename, 'w')
f.write(str(os.getpid()))
f.close()
except IO_ERRORS:
pass
def unlinkPidFile(self):
if not self.cfg.zserver_read_only_mode:
try:
os.unlink(self.cfg.pid_filename)
except OSError:
pass
def unlinkLockFile(self):
if not self.cfg.zserver_read_only_mode and hasattr(self, 'lockfile'):
try:
self.lockfile.close()
os.unlink(self.cfg.lock_filename)
except OSError:
pass
class WindowsZopeStarter(WindowsWSGIStarter, ZopeStarter):
pass
class UnixZopeStarter(UnixWSGIStarter, ZopeStarter):
pass
def dropPrivileges(cfg):
# Drop root privileges if we have them and we're on a posix platform.
# This needs to be a function so it may be used outside of Zope
# appserver startup (e.g. from zopectl debug)
if os.name != 'posix':
return
if os.getuid() != 0:
return
import pwd
effective_user = cfg.effective_user
if effective_user is None:
msg = ('A user was not specified to setuid to; fix this to '
'start as root (change the effective-user directive '
'in zope.conf)')
logger.critical(msg)
raise ZConfig.ConfigurationError(msg)
try:
uid = int(effective_user)
except ValueError:
try:
pwrec = pwd.getpwnam(effective_user)
except KeyError:
msg = "Can't find username %r" % effective_user
logger.error(msg)
raise ZConfig.ConfigurationError(msg)
uid = pwrec[2]
else:
try:
pwrec = pwd.getpwuid(uid)
except KeyError:
msg = "Can't find uid %r" % uid
logger.error(msg)
raise ZConfig.ConfigurationError(msg)
gid = pwrec[3]
if uid == 0:
msg = 'Cannot start Zope with the effective user as the root user'
logger.error(msg)
raise ZConfig.ConfigurationError(msg)
try:
from os import initgroups
initgroups(effective_user, gid)
os.setgid(gid)
except OSError:
logger.exception('Could not set group id of effective user')
os.setuid(uid)
logger.info('Set effective user to "%s"' % effective_user)
return 1 # for unit testing purposes
def _name2Ips(host, isIp_=compile(r'(\d+\.){3}').match):
'''map a name *host* to the sequence of its ip addresses;
use *host* itself (as sequence) if it already is an ip address.
Thus, if only a specific interface on a host is trusted,
identify it by its ip (and not the host name).
'''
if isIp_(host):
return [host]
return gethostbyaddr(host)[2]
src/Zope2/Startup/handlers.py
View file @ 8e149c4b
......@@ -130,7 +130,7 @@ def root_handler(cfg):
if cfg.trusted_proxies:
mapped = []
for name in cfg.trusted_proxies:
mapped.extend(_name2Ips(name))
mapped.extend(_name_to_ips(name))
config.TRUSTED_PROXIES = tuple(mapped)
from ZPublisher import HTTPRequest
......@@ -150,13 +150,13 @@ def handleConfig(cfg, multihandler):
return multihandler(handlers)
def _name2Ips(host, isIp_=re.compile(r'(\d+\.){3}').match):
def _name_to_ips(host, _is_ip=re.compile(r'(\d+\.){3}').match):
"""Map a name *host* to the sequence of its ip addresses.
use *host* itself (as sequence) if it already is an ip address.
Thus, if only a specific interface on a host is trusted,
identify it by its ip (and not the host name).
"""
if isIp_(host):
if _is_ip(host):
return [host]
return gethostbyaddr(host)[2]
src/Zope2/Startup/misc/TextBlockFormatter.py deleted 100644 → 0
View file @ efe61189
##############################################################################
#
# 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
#
##############################################################################
import string, math
def format(text, max_width=80, indent=0, trailing_lines_indent_more=0):
text = string.expandtabs(string.replace(text, '\r', ''))
lines = string.split(text, '\n')
aggregate = []
for line in lines:
if len(line) <= max_width-1:
# this line is short enough to output
aggregate.append(line)
else:
lines = splitlongline(line, max_width)
aggregate.extend(lines)
out = []
i = 0
for line in aggregate:
spaces = ' ' * indent
if i != 0 and trailing_lines_indent_more:
spaces = spaces + (' ' * trailing_lines_indent_more)
out.append('%s%s' % (spaces, line))
i = i + 1
return string.join(out, '\n')
def splitword(word, max_width=80, linepos=0):
# some lines may have single words that exceed the max_width
# We want to break apart long words into as many chunks as necessary
if len(word) <= max_width:
return [word]
first_chunk_len = max_width-1-linepos
firstchunk = word[:first_chunk_len]
word = word[first_chunk_len:]
numchunks = int(math.ceil(len(word) / float(max_width-1)))
index = 0
tmp = [firstchunk]
for chunknum in range(numchunks):
chunk = word[index:index+max_width-1]
tmp.append(chunk)
index = index + max_width-1
return tmp
def splitlongline(line, max_width=80):
# split a "long" line defined by max_width into a list of lines
line = string.strip(line)
words = string.split(line, ' ')
wordnum = 0
# iterate over all the words in the line, extending the word list
# necessary for too-long words
aggregate = []
linelen = 0
wordnum = 0
while words:
word = words.pop(0)
if not word: continue
if len(word) > max_width:
new_words = splitword(word, max_width, linelen)
word = new_words[0]
for new_word in new_words[1:]:
words.insert(wordnum, new_word)
wordnum = wordnum + 1
if words:
next_word = words[0]
else:
next_word = None
if next_word is None:
aggregate.append(word)
wordnum = wordnum + 1
continue
maybe_len = linelen + len(word) + len(next_word)
if maybe_len >= max_width-1:
aggregate.append(word)
aggregate.append(None)
linelen = 0
else:
aggregate.append(word)
linelen = linelen + len(word) + 1
wordnum = wordnum + 1
s = ""
last = None
for item in aggregate:
if item is None:
s = '%s\n' % s
elif last is None:
s = '%s%s' % (s, item)
else:
s = '%s %s' % (s, item)
last = item
return string.split(s, '\n')
long = """
To turn a component into a product you must fulfill many contracts. For the most part these contracts are not yet defined in terms of interfaces. Instead you must subclass from base classes that implement the contracts. This makes building products confusing, and this is an area that we are actively working on improving. Hereisalonglinethatshouldbecaughtandbrokenupbytheformatter,hopefullyitllgetsplitupwhenirunitthroughthere."""
long2 = """
Hereisalonglinethatshouldbecaughtandbrokenupbytheformatter,hopefullyitllgetsplitupwhenirunitthroughthere."""
long3 = """
To turn a component into a product you must fulfill many contracts. For the most part these contracts are not yet defined in terms of interfaces.
Instead you must subclass from base classes that implement the contracts. This makes building products confusing, and this is an area that we are
actively working on improving. Hereisalonglinethatshouldbecaughtandbrokenupbytheformatter,hopefullyitllgetsplitupwhenirunitthroughthere."""
if __name__ == '__main__':
print format(long, 60, 10)
print format(long)
print format(long2, 60, 10)
print format(long2)
print format(long3, 60, 10)
print format(long3)
src/Zope2/Startup/misc/lock_file.py
View file @ 8e149c4b
......@@ -12,29 +12,9 @@
#
##############################################################################
"""Utility function for file locking.
from zope.deferredimport import deprecated
This module provides a platform-specific function which uses the
best-available strategy for locking a file object.
"""
try:
import fcntl
except ImportError:
# Try windows-specific code:
try:
# We expect this module to exist, but the LockFile function may not.
from ZODB.winlock import LockFile
except ImportError:
# we don't understand any kind of locking, forget it
def lock_file(file):
pass
else:
# Windows
def lock_file(file):
un = file.fileno()
LockFile(un, 0, 0, 1, 0) # just lock the first byte, who cares
else:
# Unix-specific locking:
def lock_file(file):
fcntl.flock(file, fcntl.LOCK_EX | fcntl.LOCK_NB)
# BBB Zope 5.0
deprecated(
'Please import from ZServer.Zope2.Startup.utils',
lock_file='ZServer.Zope2.Startup.utils:lock_file')
src/Zope2/Startup/misc/zpasswd.py deleted 100644 → 0
View file @ efe61189
#!/usr/bin/env python
##############################################################################
#
# Copyright (c) 2002 Zope Foundation and Contributors.
#
# 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
#
##############################################################################
"""Zope user bootstrap system"""
import sys, sha, binascii, random, getopt, getpass, os
try:
from crypt import crypt
except ImportError:
crypt = None
def generate_salt():
"""Generate a salt value for the crypt function."""
salt_choices = ("ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"abcdefghijklmnopqrstuvwxyz"
"0123456789./")
return random.choice(salt_choices)+random.choice(salt_choices)
def generate_passwd(password, encoding):
encoding=encoding.upper()
if encoding == 'SHA':
pw = '{SHA}' + binascii.b2a_base64(sha.new(password).digest())[:-1]
elif encoding == 'CRYPT':
pw = '{CRYPT}' + crypt(password, generate_salt())
elif encoding == 'CLEARTEXT':
pw = password
return pw
def write_generated_password(home, ac_path, username):
pw_choices = ("ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"abcdefghijklmnopqrstuvwxyz"
"0123456789!")
acfile=open(ac_path, 'w')
pw = ''
for i in range(8):
pw = pw + random.choice(pw_choices)
acfile.write('%s:%s' % (username, generate_passwd(pw, 'SHA')))
acfile.close()
os.system('chmod 644 %s' % ac_path)
return pw
def write_access(home, user='', group=''):
ac_path=os.path.join(home, 'access')
if not os.path.exists(ac_path):
print '-'*78
print 'creating default access file'
pw = write_generated_password(home, ac_path, 'emergency')
print """Note:
The emergency user name and password are 'emergency'
and '%s'.
You can change the emergency name and password with the
zpasswd script. To find out more, type:
%s zpasswd.py
""" % (pw, sys.executable)
import do; do.ch(ac_path, user, group)
def write_inituser(home, user='', group=''):
ac_path=os.path.join(home, 'inituser')
if not os.path.exists(ac_path):
print '-'*78
print 'creating default inituser file'
pw = write_generated_password(home, ac_path, 'admin')
print """Note:
The initial user name and password are 'admin'
and '%s'.
You can change the name and password through the web
interface or using the 'zpasswd.py' script.
""" % pw
import do; do.ch(ac_path, user, group)
class CommandLineError(Exception):
pass
def main(argv):
short_options = ':u:p:e:d:'
long_options = ['username=',
'password=',
'encoding=',
'domains=']
usage = """Usage: %s [options] filename
If this program is called without command-line options, it will prompt
for all necessary information. The available options are:
-u / --username=
Set the username to be used for the initial user or the emergency user
-p / --password=
Set the password
-e / --encoding=
Set the encryption/encoding rules. Defaults to SHA-1. OPTIONAL
-d / --domains=
Set the domain names that the user user can log in from. Defaults to
any. OPTIONAL.
Filename is required and should be the name of the file to store the
information in (usually "inituser" or "access").
Copyright (C) 1999, 2000 Digital Creations, Inc.
""" % argv[0]
try:
if len(argv) < 2:
raise CommandLineError, "Not enough arguments!"
optlist, args = getopt.getopt(sys.argv[1:], short_options, long_options)
if len(args) != 1:
raise CommandLineError, "Only one filename allowed!"
access_file = open(args[0], 'w')
if len(optlist) > 0:
# Set the sane defaults
username = ''
encoding = 'SHA'
domains = ''
for opt in optlist:
if (opt[0] == '-u') or (opt[0] == '--username'):
username = opt[1]
elif (opt[0] == '-p') or (opt[0] == '--password'):
password = opt[1]
elif (opt[0] == '-e') or (opt[0] == '--encoding'):
encoding = opt[1]
elif (opt[0] == '-d') or (opt[0] == '--domains'):
domains = ":" + opt[1]
# Verify that we got what we need
if not username or not password:
raise CommandLineError, "Must specify username and password."
access_file.write(username + ':' +
generate_passwd(password, encoding) +
domains)
else:
# Run through the prompts
while 1:
username = raw_input("Username: ")
if username != '':
break
while 1:
password = getpass.getpass("Password: ")
verify = getpass.getpass("Verify password: ")
if verify == password:
break
else:
password = verify = ''
print "Password mismatch, please try again..."
while 1:
print """
Please choose a format from:
SHA - SHA-1 hashed password (default)
CRYPT - UNIX-style crypt password
CLEARTEXT - no protection
"""
encoding = raw_input("Encoding: ")
if encoding == '':
encoding = 'SHA'
break
if encoding.upper() in ['SHA', 'CRYPT', 'CLEARTEXT']:
break
domains = raw_input("Domain restrictions: ")
if domains: domains = ":" + domains
access_file.write(username + ":" +
generate_passwd(password, encoding) +
domains)
except CommandLineError, v:
sys.stderr.write(usage)
sys.stderr.write("\n\n%s" % v)
sys.exit(1)
# If called from the command line
if __name__=='__main__':
main(sys.argv)
src/Zope2/Startup/starter.py 0 → 100644
View file @ 8e149c4b
##############################################################################
#
# Copyright (c) 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.
#
##############################################################################
import logging
import sys
import re
from socket import gethostbyaddr
from ZConfig import ConfigurationError
from zope.event import notify
from zope.processlifetime import ProcessStarting
logger = logging.getLogger("Zope")
class WSGIStarter(object):
"""This is a class which starts Zope as a WSGI app."""
wsgi = True
def prepare(self):
self.setupLocale()
self.setupSecurityOptions()
self.setupPublisher()
self.setupInterpreter()
self.startZope()
from App.config import getConfiguration
config = getConfiguration() # NOQA
notify(ProcessStarting())
logger.info('Ready to handle requests')
def setConfiguration(self, cfg):
self.cfg = cfg
def setupInterpreter(self):
# make changes to the python interpreter environment
sys.setcheckinterval(self.cfg.python_check_interval)
def setupLocale(self):
# set a locale if one has been specified in the config
if not self.cfg.locale:
return
# workaround to allow unicode encoding conversions in DTML
import codecs
dummy = codecs.lookup('utf-8') # NOQA
locale_id = self.cfg.locale
if locale_id is not None:
try:
import locale
except:
raise ConfigurationError(
'The locale module could not be imported.\n'
'To use localization options, you must ensure\n'
'that the locale module is compiled into your\n'
'Python installation.')
try:
locale.setlocale(locale.LC_ALL, locale_id)
except:
raise ConfigurationError(
'The specified locale "%s" is not supported by your'
'system.\nSee your operating system documentation for '
'more\ninformation on locale support.' % locale_id)
def setupPublisher(self):
import ZPublisher.HTTPRequest
import ZPublisher.Publish
ZPublisher.Publish.set_default_debug_mode(self.cfg.debug_mode)
ZPublisher.Publish.set_default_authentication_realm(
self.cfg.http_realm)
if self.cfg.trusted_proxies:
mapped = []
for name in self.cfg.trusted_proxies:
mapped.extend(_name_to_ips(name))
ZPublisher.HTTPRequest.trusted_proxies = tuple(mapped)
def setupSecurityOptions(self):
import AccessControl
AccessControl.setImplementation(
self.cfg.security_policy_implementation)
AccessControl.setDefaultBehaviors(
not self.cfg.skip_ownership_checking,
not self.cfg.skip_authentication_checking,
self.cfg.verbose_security)
def startZope(self):
# Import Zope
import Zope2
Zope2.startup()
def _name_to_ips(host, _is_ip=re.compile(r'(\d+\.){3}').match):
'''map a name *host* to the sequence of its ip addresses;
use *host* itself (as sequence) if it already is an ip address.
Thus, if only a specific interface on a host is trusted,
identify it by its ip (and not the host name).
'''
if _is_ip(host):
return [host]
return gethostbyaddr(host)[2]
src/Zope2/Startup/tests/testStarter.py
View file @ 8e149c4b
......@@ -11,84 +11,46 @@
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
""" Tests of the ZopeStarter class """
import cStringIO
import errno
import logging
import os
import random
import shutil
import sys
import tempfile
import unittest
import ZConfig
from ZConfig.components.logger.tests.test_logger import LoggingTestHelper
from App.config import getConfiguration, setConfiguration
import Products
from Zope2.Startup import get_starter
from Zope2.Startup.options import ZopeOptions
TEMPNAME = tempfile.mktemp()
TEMPPRODUCTS = os.path.join(TEMPNAME, "Products")
_SCHEMA = {}
LIFETIME = True
try:
import Lifetime # NOQA
except ImportError:
LIFETIME = False
_SCHEMA = None
def getSchema(schemafile):
global _SCHEMA
if schemafile not in _SCHEMA:
if _SCHEMA is None:
opts = ZopeOptions()
opts.schemafile = schemafile
opts.load_schema()
_SCHEMA[schemafile] = opts.schema
return _SCHEMA[schemafile]
# try to preserve logging state so we don't screw up other unit tests
# that come later
_SCHEMA = opts.schema
return _SCHEMA
logger_states = {}
for name in (None, 'trace', 'access'):
logger = logging.getLogger(name)
logger_states[name] = {'level': logger.level,
'propagate': logger.propagate,
'handlers': logger.handlers,
'filters': logger.filters}
class WSGIStarterTestCase(unittest.TestCase):
class BaseTestCase(LoggingTestHelper):
@property
def schema(self):
return getSchema('wsgischema.xml')
def setUp(self):
LoggingTestHelper.setUp(self)
self.TEMPNAME = tempfile.mktemp()
def tearDown(self):
try:
os.rmdir(TEMPPRODUCTS)
os.rmdir(TEMPNAME)
except:
pass
Products.__path__ = [d for d in Products.__path__
if os.path.exists(d)]
LoggingTestHelper.tearDown(self)
shutil.rmtree(self.TEMPNAME)
# reset logger states
for name in (None, 'access', 'trace'):
logger = logging.getLogger(name)
logger.__dict__.update(logger_states[name])
def _clearHandlers(self):
from ZConfig.components.logger import loghandler
del loghandler._reopenable_handlers[:]
def get_starter(self, conf, wsgi=False):
starter = get_starter(wsgi=wsgi)
def get_starter(self, conf):
starter = get_starter(wsgi=True)
starter.setConfiguration(conf)
return starter
......@@ -96,27 +58,18 @@ class BaseTestCase(LoggingTestHelper):
# We have to create a directory of our own since the existence
# of the directory is checked. This handles this in a
# platform-independent way.
schema = self.schema
sio = cStringIO.StringIO(
text.replace("<<INSTANCE_HOME>>", TEMPNAME))
text.replace("<<INSTANCE_HOME>>", self.TEMPNAME))
try:
os.mkdir(TEMPNAME)
os.mkdir(TEMPPRODUCTS)
os.mkdir(self.TEMPNAME)
except OSError as why:
if why == 17:
# already exists
pass
conf, self.handler = ZConfig.loadConfigFile(schema, sio)
self.assertEqual(conf.instancehome, TEMPNAME)
conf, self.handler = ZConfig.loadConfigFile(self.schema, sio)
self.assertEqual(conf.instancehome, self.TEMPNAME)
return conf
class WSGIStarterTestCase(BaseTestCase, unittest.TestCase):
@property
def schema(self):
return getSchema('wsgischema.xml')
def testSetupLocale(self):
# XXX this almost certainly won't work on all systems
import locale
......@@ -131,105 +84,14 @@ class WSGIStarterTestCase(BaseTestCase, unittest.TestCase):
'The specified locale "en_GB" is not supported'):
return
raise
starter = self.get_starter(conf, wsgi=True)
starter = self.get_starter(conf)
starter.setupLocale()
self.assertEqual(tuple(locale.getlocale()), ('en_GB', 'ISO8859-1'))
finally:
# reset to system-defined locale
locale.setlocale(locale.LC_ALL, '')
def testSetupStartupHandler(self):
if sys.platform[:3].lower() == "win":
return
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
debug-mode on
<eventlog>
level info
<logfile>
path <<INSTANCE_HOME>>/event.log
level info
</logfile>
<logfile>
path <<INSTANCE_HOME>>/event2.log
level blather
</logfile>
</eventlog>""")
starter = self.get_starter(conf, wsgi=True)
starter.setupInitialLogging()
# startup handler should take on the level of the event log handler
# with the lowest level
logger = starter.event_logger
self.assertEqual(starter.startup_handler.level, 15) # 15 is BLATHER
self.assert_(starter.startup_handler in logger.handlers)
self.assertEqual(logger.level, 15)
# We expect a debug handler and the startup handler:
self.assertEqual(len(logger.handlers), 2)
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
debug-mode off
<eventlog>
level info
<logfile>
path <<INSTANCE_HOME>>/event.log
level info
</logfile>
</eventlog>""")
starter = self.get_starter(conf, wsgi=True)
starter.setupInitialLogging()
def testSetupConfiguredLoggers(self):
if sys.platform[:3].lower() == "win":
return
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
debug-mode off
<eventlog>
level info
<logfile>
path <<INSTANCE_HOME>>/event.log
level info
</logfile>
</eventlog>
<logger access>
level info
<logfile>
path <<INSTANCE_HOME>>/Z2.log
</logfile>
</logger>
<logger trace>
level info
<logfile>
path <<INSTANCE_HOME>>/trace.log
</logfile>
</logger>
""")
try:
starter = self.get_starter(conf, wsgi=True)
starter.setupInitialLogging()
zope_logger = logging.getLogger("Zope")
zope_logger.info('hello')
starter.setupFinalLogging()
logger = logging.getLogger()
self.assertEqual(logger.level, logging.INFO)
l = open(os.path.join(TEMPNAME, 'event.log')).read()
self.assertTrue(l.find('hello') > -1)
self.assertTrue(
os.path.exists(os.path.join(TEMPNAME, 'Z2.log')))
self.assertTrue(
os.path.exists(os.path.join(TEMPNAME, 'trace.log')))
finally:
for name in ('event.log', 'Z2.log', 'trace.log'):
try:
os.unlink(os.path.join(TEMPNAME, name))
except:
pass
self._clearHandlers()
def testConfigureInterpreter(self):
import sys
oldcheckinterval = sys.getcheckinterval()
newcheckinterval = oldcheckinterval + 1
conf = self.load_config_text("""
......@@ -237,177 +99,8 @@ class WSGIStarterTestCase(BaseTestCase, unittest.TestCase):
python-check-interval %d
""" % newcheckinterval)
try:
starter = self.get_starter(conf, wsgi=True)
starter = self.get_starter(conf)
starter.setupInterpreter()
self.assertEqual(sys.getcheckinterval(), newcheckinterval)
finally:
sys.setcheckinterval(oldcheckinterval)
if LIFETIME:
class ZopeStarterTestCase(BaseTestCase, unittest.TestCase):
@property
def schema(self):
return getSchema('zopeschema.xml')
def testDropPrivileges(self):
# somewhat incomplete because we we're never running as root
# when we test, but we test as much as we can
if os.name != 'posix':
return
_old_getuid = os.getuid
def _return0():
return 0
def make_starter(conf):
# remove the debug handler, since we don't want junk on
# stderr for the tests
starter = self.get_starter(conf, wsgi=False)
starter.event_logger.removeHandler(starter.debug_handler)
return starter
try:
os.getuid = _return0
# no effective user
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>""")
starter = make_starter(conf)
self.assertRaises(ZConfig.ConfigurationError,
starter.dropPrivileges)
# cant find user in passwd database
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
effective-user n0sucHuS3r""")
starter = make_starter(conf)
self.assertRaises(ZConfig.ConfigurationError,
starter.dropPrivileges)
# can't specify '0' as effective user
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
effective-user 0""")
starter = make_starter(conf)
self.assertRaises(ZConfig.ConfigurationError,
starter.dropPrivileges)
# setuid to test runner's uid
runnerid = _old_getuid()
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
effective-user %s""" % runnerid)
starter = make_starter(conf)
finished = starter.dropPrivileges()
self.assertTrue(finished)
finally:
os.getuid = _old_getuid
def testMakeLockFile(self):
# put something in the way (it should be deleted)
name = os.path.join(TEMPNAME, 'lock')
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
lock-filename %s""" % name
)
f = open(name, 'ab')
# On Windows, the first byte of the file is locked solid, and even
# we (this process) can't read from it via a file object other
# than the one passed to lock_file. So we put a blank
# in the test value first, so we can skip over it later. Also,
# because .seek(1) isn't well-defined for files opened in text
# mode, we open the file in binary mode (above and below).
f.write(' hello')
f.close()
try:
starter = self.get_starter(conf, wsgi=False)
starter.makeLockFile()
f = open(name, 'rb')
f.seek(1) # skip over the locked byte
guts = f.read()
f.close()
self.assertFalse(guts.find('hello') > -1)
finally:
starter.unlinkLockFile()
self.assertFalse(os.path.exists(name))
def testMakePidFile(self):
# put something in the way (it should be deleted)
name = os.path.join(TEMPNAME, 'pid')
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
pid-filename %s""" % name
)
f = open(name, 'a')
f.write('hello')
f.close()
try:
starter = self.get_starter(conf, wsgi=False)
starter.makePidFile()
self.assertFalse(open(name).read().find('hello') > -1)
finally:
starter.unlinkPidFile()
self.assertFalse(os.path.exists(name))
def testSetupZServerThreads(self):
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
zserver-threads 10""")
starter = self.get_starter(conf, wsgi=False)
starter.setupZServer()
from Zope2.Startup.config import ZSERVER_THREADS
self.assertEqual(ZSERVER_THREADS, 10)
def testSetupServers(self):
# We generate a random port number to test against, so that
# multiple test runs of this at the same time can succeed
port = random.randint(10000, 50000)
conf = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
<http-server>
address %(http)s
</http-server>
<ftp-server>
address %(ftp)s
</ftp-server>""" % dict(http=port, ftp=port + 1)
)
starter = self.get_starter(conf, wsgi=False)
# do the job the 'handler' would have done (call prepare)
for server in conf.servers:
server.prepare('', None, 'Zope2', {}, None)
try:
starter.setupServers()
import ZServer
self.assertEqual(conf.servers[0].__class__,
ZServer.HTTPServer.zhttp_server)
self.assertEqual(conf.servers[1].__class__,
ZServer.FTPServer)
finally:
del conf.servers # should release servers
pass
def testZopeRunConfigure(self):
old_config = getConfiguration()
try:
os.mkdir(TEMPNAME)
os.mkdir(TEMPPRODUCTS)
except OSError as why:
if why == errno.EEXIST:
# already exists
pass
old_argv = sys.argv
sys.argv = [sys.argv[0]]
try:
fname = os.path.join(TEMPNAME, 'zope.conf')
from Zope2 import configure
f = open(fname, 'w')
f.write('instancehome %s\nzserver-threads 100\n' % TEMPNAME)
f.flush()
f.close()
configure(fname)
new_config = getConfiguration()
self.assertEqual(new_config.zserver_threads, 100)
finally:
sys.argv = old_argv
try:
os.unlink(fname)
except Exception:
pass
setConfiguration(old_config)
src/Zope2/Startup/tests/test_schema.py
View file @ 8e149c4b
......@@ -20,19 +20,10 @@ import unittest
import ZConfig
import Products
from Zope2.Startup import datatypes
from Zope2.Startup.options import ZopeOptions
LIFETIME = True
try:
import Lifetime # NOQA
except ImportError:
LIFETIME = False
_SCHEMA = {}
TEMPNAME = tempfile.mktemp()
TEMPPRODUCTS = os.path.join(TEMPNAME, "Products")
TEMPVAR = os.path.join(TEMPNAME, "var")
......@@ -99,9 +90,9 @@ class WSGIStartupTestCase(unittest.TestCase):
<filestorage>
path <<INSTANCE_HOME>>/var/Data.fs
</filestorage>
mount-point /
cache-size 5000
pool-size 7
mount-point /
cache-size 5000
pool-size 7
</zodb_db>
""")
self.assertEqual(conf.databases[0].config.cache_size, 5000)
......@@ -130,105 +121,3 @@ class WSGIStartupTestCase(unittest.TestCase):
default-zpublisher-encoding iso-8859-15
""")
self.assertEqual(conf.default_zpublisher_encoding, 'iso-8859-15')
if LIFETIME:
class ZServerStartupTestCase(unittest.TestCase):
def tearDown(self):
Products.__path__ = [d for d in Products.__path__
if os.path.exists(d)]
@property
def schema(self):
return getSchema('zopeschema.xml')
def load_config_text(self, text):
# We have to create a directory of our own since the existence
# of the directory is checked. This handles this in a
# platform-independent way.
schema = self.schema
sio = cStringIO.StringIO(
text.replace("<<INSTANCE_HOME>>", TEMPNAME))
os.mkdir(TEMPNAME)
os.mkdir(TEMPPRODUCTS)
os.mkdir(TEMPVAR)
try:
conf, handler = ZConfig.loadConfigFile(schema, sio)
finally:
os.rmdir(TEMPPRODUCTS)
os.rmdir(TEMPVAR)
os.rmdir(TEMPNAME)
self.assertEqual(conf.instancehome, TEMPNAME)
return conf, handler
def test_cgi_environment(self):
conf, handler = self.load_config_text("""\
# instancehome is here since it's required
instancehome <<INSTANCE_HOME>>
<cgi-environment>
HEADER value
ANOTHER value2
</cgi-environment>
""")
items = conf.cgi_environment.items()
items.sort()
self.assertEqual(
items, [("ANOTHER", "value2"), ("HEADER", "value")])
def test_ms_public_header(self):
from Zope2.Startup import config
from Zope2.Startup.handlers import handleConfig
default_setting = config.ZSERVER_ENABLE_MS_PUBLIC_HEADER
try:
conf, handler = self.load_config_text("""\
instancehome <<INSTANCE_HOME>>
enable-ms-public-header true
""")
handleConfig(None, handler)
self.assertTrue(config.ZSERVER_ENABLE_MS_PUBLIC_HEADER)
conf, handler = self.load_config_text("""\
instancehome <<INSTANCE_HOME>>
enable-ms-public-header false
""")
handleConfig(None, handler)
self.assertFalse(config.ZSERVER_ENABLE_MS_PUBLIC_HEADER)
finally:
config.ZSERVER_ENABLE_MS_PUBLIC_HEADER = default_setting
def test_path(self):
p1 = tempfile.mktemp()
p2 = tempfile.mktemp()
try:
os.mkdir(p1)
os.mkdir(p2)
conf, handler = self.load_config_text("""\
# instancehome is here since it's required
instancehome <<INSTANCE_HOME>>
path %s
path %s
""" % (p1, p2))
items = conf.path
self.assertEqual(items, [p1, p2])
finally:
if os.path.exists(p1):
os.rmdir(p1)
if os.path.exists(p2):
os.rmdir(p2)
def test_access_and_trace_logs(self):
fn = tempfile.mktemp()
conf, handler = self.load_config_text("""
instancehome <<INSTANCE_HOME>>
<logger access>
<logfile>
path %s
</logfile>
</logger>
""" % fn)
self.assert_(isinstance(conf.access, datatypes.LoggerFactory))
self.assertEqual(conf.access.name, "access")
self.assertEqual(conf.access.handler_factories[0].section.path, fn)
self.assert_(conf.trace is None)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7