Skip to content

Z
Zope
Commit 50c195d3 authored by Jim Fulton's avatar Jim Fulton
Browse files
Options

Updated code to: 

  - Provide a first cut authentication.authorization scheme
  - Fix several bugs
  - Provide better error messages
  - Provide automagic insertion of base
  - Support Fast CGI module publisher.
parent 0a9dc8e1
Hide whitespace changes
Inline Side-by-side
Showing with 340 additions and 133 deletions
lib/python/ZPublisher/Publish.py
View file @ 50c195d3
#! /usr/local/bin/python
import sys, os, string, types, cgi
from CGIResponse import Response, ExceptionResponse
class output_file:
'''Generic output abstraction output for CGI resource output.
This may be a thin layer around stdout, or it may be something
very different.'''
called=None
import sys, os, string, types, newcgi, CGIResponse, regex
class ModulePublisher:
def html(self,title,body):
return ("<html>\n"
"<head>\n<title>%s</title>\n</head>\n"
"<body>\n%s\n</body>\n"
"</html>\n" % (title,body))
def notFoundError(self):
raise 'NotFound',self.html(
"Resource not found",
"Sorry, the requested document does not exist.")
forbiddenError=notFoundError # If a resource is forbidden,
# why reveal that it exists?
def badRequestError(self,name):
if regex.match('^[A-Z_0-9]+$',name) >= 0:
raise 'InternalError', self.html(
"Internal Error",
"Sorry, an internal error occurred in this resource.")
raise 'BadRequest',self.html(
"Invalid request",
"The parameter, %s, was ommitted from the request." % name)
def forbiddenError(self):
raise 'NotFound',self.html(
"Resource not found",
"Sorry, the requested document does not exist.")
def write(self,s):
self.called=1
self.write=sys.stdout.write
if string.lower(string.strip(s)[:6]) == '<html>':
print 'content-type: text/html\n'
elif string.lower(s)[:13]!='content-type:':
print 'content-type: text/plain\n'
self.write(s)
def _publish_module(module_name, published='web_objects'):
dict={}
exec 'import %s' % module_name in dict
theModule=dict=dict[module_name]
if hasattr(dict,published):
dict=getattr(dict,published)
else:
dict=dict.__dict__
published=None
def env(key,d=os.environ):
try: return d[key]
def env(self,key):
try: return self.environ[key]
except: return ''
query=cgi.parse() or {}
path=(string.strip(env('PATH_INFO')) or '/')[1:]
path=string.splitfields(path,'/')
while path and not path[0]: path = path[1:]
if not path and dict['__doc__']: function=theModule
else:
if not path: path = ['main']
function,path=dict[path[0]], path[1:]
if not (published or function.__doc__):
raise 'Forbidden',function
p=''
while path:
p,path=path[0], path[1:]
if p:
try: f=getattr(function,p)
def document(self,o,response):
if type(o) is not types.StringType: o=o.__doc__
response.setBody(
self.html('Documentation for' +
((self.env('PATH_INFO') or
('/'+self.module_name))[1:]),
'<pre>\n%s\n</pre>' % o)
)
return response
def validate(self,object,module,response):
if hasattr(object,'__allow_groups__'):
groups=object.__allow_groups__
try: realm=module.__realm__
except: self.forbiddenError()
try:
user=realm.user(self.env("HTTP_AUTHORIZATION"))
if type(groups) is type({}):
groups=map(lambda k,d=groups: d[k], groups.keys())
for g in groups:
if g.has_key(user): return None
realm.deny()
except:
f=function[p]
function=f
if not (p=='__doc__' or function.__doc__):
raise 'Forbidden',function
try:
t,v,tb=sys.exc_type, sys.exc_value,sys.exc_traceback
auth,v=v
except: pass
if t == 'Unauthorized':
response['WWW-authenticate']=auth
raise 'Unauthorized', v
self.forbiddenError()
def publish(self, module_name, response, published='web_objects',
imported_modules={}, module_dicts={}):
if module_name[-4:]=='.cgi': module_name=module_name[:-4]
self.module_name=module_name
response.setBase(self.base)
dict=imported_modules
try:
theModule, dict, published = module_dicts[module_name]
except:
exec 'import %s' % module_name in dict
theModule=dict=dict[module_name]
if hasattr(dict,published):
dict=getattr(dict,published)
else:
dict=dict.__dict__
published=None
module_dicts[module_name] = theModule, dict, published
path=(string.strip(self.env('PATH_INFO')) or '/')[1:]
path=string.splitfields(path,'/')
while path and not path[0]: path = path[1:]
if not path: path = ['help']
try: function,path=dict[path[0]], path[1:]
except KeyError: self.notFoundError()
def document(o,env=env):
if type(o) is not types.StringType: o=o.__doc__
return Response(('Documentation for' +
((env('PATH_INFO') or '/main')[1:]),
'<pre>\n%s\n</pre>' % o))
if(p=='__doc__' or env('METHOD_NAME') == 'HEAD'):
return document(function)
f=function
if type(f) is types.ClassType:
if hasattr(f,'__init__'):
f=f.__init__.im_func
if not (published or function.__doc__): self.forbiddenError()
self.validate(function,theModule,response)
p=''
while path:
p,path=path[0], path[1:]
if p:
try: f=getattr(function,p)
except AttributeError:
try: f=function[p]
except TypeError:
if not path and p=='help':
p, f = '__doc__', self.document(function,response)
else:
self.notFoundError()
function=f
if not (p=='__doc__' or function.__doc__):
raise 'Forbidden',function
self.validate(function,theModule,response)
f=function
if type(f) is types.ClassType:
if hasattr(f,'__init__'):
f=f.__init__
else:
def ff(): pass
f=ff
if type(f) is types.MethodType:
defaults=f.im_func.func_defaults
names=(f.im_func.
func_code.co_varnames[1:f.im_func.func_code.co_argcount])
elif type(f) is types.FunctionType:
defaults=f.func_defaults
names=f.func_code.co_varnames[:f.func_code.co_argcount]
else:
def ff(): pass
f=ff
if type(f) is types.MethodType:
defaults=f.im_func.func_defaults
names=(f.im_func.
func_code.co_varnames[1:f.im_func.func_code.co_argcount])
elif type(f) is types.FunctionType:
defaults=f.func_defaults
names=f.func_code.co_varnames[:f.func_code.co_argcount]
else:
return document(function)
environ=os.environ
for key in environ.keys():
query[key]=[environ[key]]
out=output_file()
query['OUTPUT_FILE']=[out]
last_name=len(names) - (len(defaults or []))
for name in names[:last_name]:
if not query.has_key(name): raise 'BadRequest', query
args={}
for name in names:
if query.has_key(name):
q=query[name]
if len(q) == 1: q=q[0]
args[name]=q
if args: result=apply(function,(),args)
else: result=function()
if out.called: # The function wrote output
if result: result=str(result)
else: result=''
return result
return response.setBody(function)
if type(result) is types.InstanceType and result.__class__ is Response:
return result
return Response(result)
def publish_module(module_name, published='web_objects'):
query=self.get_data()
query['RESPONSE']=response
last_name=len(names) - (len(defaults or []))
for name in names[:last_name]:
if not query.has_key(name):
self.badRequestError(name)
args={}
for name in names:
if query.has_key(name):
q=query[name]
if type(q) is type([]) and len(q) == 1: q=q[0]
args[name]=q
if args: result=apply(function,(),args)
else: result=function()
if result and result is not response: response.setBody(result)
return response
class CGIModulePublisher(ModulePublisher):
def __init__(self,
stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr,
environ=os.environ):
self.environ=environ
self.stdin=stdin
self.stdout=stdout
self.stderr=stderr
b=string.strip(self.environ['SCRIPT_NAME'])
if b[-1]=='/': b=b[:-1]
p = string.rfind(b,'/')
if p >= 0: self.base=b[:p+1]
else: self.base=''
special_names = {
'SERVER_SOFTWARE' : 1,
'SERVER_NAME' : 1,
'GATEWAY_INTERFACE' : 1,
'SERVER_PROTOCOL' : 1,
'SERVER_PORT' : 1,
'REQUEST_METHOD' : 1,
'PATH_INFO' : 1,
'PATH_TRANSLATED' : 1,
'SCRIPT_NAME' : 1,
'QUERY_STRING' : 1,
'REMOTE_HOST' : 1,
'REMOTE_ADDR' : 1,
'AUTH_TYPE' : 1,
'REMOTE_USER' : 1,
'REMOTE_IDENT' : 1,
'CONTENT_TYPE' : 1,
'CONTENT_LENGTH' : 1,
}
def get_data(self):
query=newcgi.parse(fp=self.stdin, environ=self.environ) or {}
environ=self.environ
for key in environ.keys():
if self.special_names.has_key(key) or key[:5] == 'HTTP_':
query[key]=[environ[key]]
for cookie in string.split(self.env('HTTP_COOKIE'),';'):
try:
[key,value]=map(string.strip, string.split(cookie,'='))
if key and not query.has_key(key): query[key]=value
except: pass
return query
def publish_module(module_name, published='web_objects',
stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr,
environ=os.environ):
response=CGIResponse.Response(stdout=stdout, stderr=stderr)
try:
response = _publish_module(module_name, published)
if response: print response
publisher=CGIModulePublisher(stdin=stdout, stdin=stdout, stderr=stderr,
environ=environ)
response = publisher.publish(module_name, response, published)
except:
print ExceptionResponse()
response.exception()
if response: response=str(response)
if response: stdout.write(response)
lib/python/ZPublisher/Response.py
View file @ 50c195d3
......@@ -3,7 +3,7 @@
__doc__='''CGI Response Output formatter
$Id: Response.py,v 1.1 1996/06/17 18:57:18 jfulton Exp $'''
$Id: Response.py,v 1.2 1996/07/01 11:51:54 jfulton Exp $'''
# Copyright
#
# Copyright 1996 Digital Creations, L.C., 910 Princess Anne
......@@ -55,14 +55,23 @@ $Id: Response.py,v 1.1 1996/06/17 18:57:18 jfulton Exp $'''
# (540) 371-6909
#
# $Log: Response.py,v $
# Revision 1.2 1996/07/01 11:51:54 jfulton
# Updated code to:
#
# - Provide a first cut authentication.authorization scheme
# - Fix several bugs
# - Provide better error messages
# - Provide automagic insertion of base
# - Support Fast CGI module publisher.
#
# Revision 1.1 1996/06/17 18:57:18 jfulton
# Almost initial version.
#
#
#
__version__='$Revision: 1.1 $'[11:-2]
__version__='$Revision: 1.2 $'[11:-2]
import string, types, sys
import string, types, sys, regex
status_codes={
'ok': 200,
......@@ -116,9 +125,13 @@ status_codes={
'zerodivisionerror':500,
}
end_of_header_re=regex.compile('</head>',regex.casefold)
base_re=regex.compile('<base',regex.casefold)
class Response:
def __init__(self,body='',status=200,headers=None):
def __init__(self,body='',status=200,headers=None,
stdout=sys.stdout, stderr=sys.stderr,):
'''\
Creates a new response. In effect, the constructor calls
"self.setBody(body); self.setStatus(status); for name in
......@@ -127,7 +140,11 @@ class Response:
headers={}
self.headers=headers
self.setStatus(status)
self.base=''
self.setBody(body)
self.cookies={}
self.stdout=stdout
self.stderr=stderr
def setStatus(self,status):
'''\
......@@ -149,7 +166,13 @@ class Response:
'''\
Sets an HTTP return header "name" with value "value", clearing
the previous value set for the header, if one exists. '''
self.headers[name]=value
self.headers[string.lower(name)]=value
def __getitem__(self, name):
'Get the value of an output header'
return self.headers[name]
__setitem__=setHeader
def setBody(self, body, title=''):
'''\
......@@ -163,17 +186,37 @@ class Response:
% (str(title),str(body)))
else:
self.body=str(body)
self.insertBase()
return self
def getStatus(self):
'Returns the current HTTP status code as an integer. '
return self.status
def setBase(self,base):
'Set the base URL for the returned document.'
self.base=base
self.insertBase()
def insertBase(self):
if self.base:
body=self.body
e=end_of_header_re.search(body)
if e >= 0 and base_re.search(body) < 0:
self.body=('%s\t<base href="%s">\n%s' %
(body[:e],self.base,body[e:]))
def appendCookie(self, name, value):
'''\
Returns an HTTP header that sets a cookie on cookie-enabled
browsers with a key "name" and value "value". If a value for the
cookie has previously been set in the response object, the new
value is appended to the old one separated by a colon. '''
try:
v,expires,domain,path,secure=self.cookies[name]
except:
v,expires,domain,path,secure='','','','',''
self.cookies[name]=v+value,expires,domain,path,secure
def expireCookie(self, name):
'''\
......@@ -181,12 +224,26 @@ class Response:
corresponding to "name" on the client, if one exists. This is
accomplished by sending a new cookie with an expiration date
that has already passed. '''
self.cookies[name]='deleted','01-Jan-96 11:11:11 GMT','','',''
def setCookie(self, name, value):
def setCookie(self,name, value=None,
expires=None, domain=None, path=None, secure=None):
'''\
Returns an HTTP header that sets a cookie on cookie-enabled
browsers with a key "name" and value "value". This overwrites
any previously set value for the cookie in the Response object. '''
try: cookie=self.cookies[name]
except: cookie=('')*5
def f(a,b):
if b is not None: return b
return a
self.cookies[name]=tuple(map(f,cookie,
(value,expires,domain,path,secure)
)
)
def appendBody(self, body):
''
......@@ -213,33 +270,82 @@ class Response:
except: h=value
self.setHeader(name,h)
def isHTML(self,str):
return string.lower(string.strip(str)[:6]) == '<html>'
def _traceback(self,t,v,tb):
import traceback
return ("\n<!--\n%s\n-->" %
string.joinfields(traceback.format_exception(t,v,tb,200),'\n')
)
def exception(self):
t,v,tb=sys.exc_type, sys.exc_value,sys.exc_traceback
self.setStatus(t)
b=v
if type(b) is not type(''):
return self.setBody(
(str(t),
'Sorry, an error occurred.<p>'
+ self._traceback(t,v,tb)))
if self.isHTML(b): return self.setBody(b+self._traceback(t,v,tb))
return self.setBody((str(t),b+self._traceback(t,v,tb)))
_wrote=None
def _cookie_list(self):
cookie_list=[]
for name in self.cookies.keys():
value,expires,domain,path,secure=self.cookies[name]
cookie='set-cookie: %s=%s' % (name,value)
if expires: cookie = "%s; expires=%s" % (cookie,expires)
if domain: cookie = "%s; domain=%s" % (cookie,domain)
if path: cookie = "%s; path=%s" % (cookie,path)
if secure: cookie = cookie+'; secure'
cookie_list.append(cookie)
return cookie_list
def __str__(self):
if self._wrote: return '' # Streaming output was used.
headers=self.headers
body=self.body
if body:
if not headers.has_key('content-type'):
if len(body) > 6 and string.lower(body[:6]) == '<html>':
if self.isHTML(body):
c='text/html'
else:
c='text/plain'
self.setHeader('content-type',c)
if not headers.has_key('content-length'):
self.setHeader('content-length',len(body))
headers=map(lambda k,d=headers: "%s: %s" % (k,d[k]), headers.keys())
if body: headers[len(headers):]=['',body]
return string.joinfields(headers,'\n')
def ExceptionResponse(body="Sorry, an error occurred"):
import traceback
t,v,tb=sys.exc_type, sys.exc_value,sys.exc_traceback
body=("%s<p>\n<!--\n%s\n-->" %
(body,
string.joinfields(traceback.format_exception(t,v,tb,200),'\n')
))
return Response(('Error',body),t)
# return Response('',t)
headersl=map(lambda k,d=headers: "%s: %s" % (k,d[k]), headers.keys())
if self.cookies:
headersl=headersl+self._cookie_list()
if body: headersl[len(headersl):]=['',body]
return string.joinfields(headersl,'\n')
def write(self,data):
self.body=self.body+data
if end_of_header_re.search(self.body) >= 0:
try: del self.headers['content-length']
except: pass
self.insertBase()
body=self.body
self.body=''
self.write=write=self.stdout.write
write(str(self))
self._wrote=1
write('\n\n')
write(body)
def ExceptionResponse():
return Response().exception()
def main():
print Response('hello world')
......
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