Return to IntranetUserFolder.py CVS log

Up to [Repository] / IntranetUserFolder

updated to work with Zope 2.8
added small check for fake forwarded-for header

"""User Folder Extension, tests now also ip number of the host where the original call comes from in case of redirects"""

import Globals
from AccessControl.User import UserFolder
from Globals import MessageDialog
import zLOG
import re

class IntranetUserFolder(UserFolder):
	"""User folder for Intranet"""
	_domain_auth_mode=1 # Identification via domain
	meta_type="IntranetUserFolder"
	
	def authenticate(self, name, password, request):
		"""modified authenticate to use domainspecmath below"""
		#zLOG.LOG('IntranetUserFolder',zLOG.INFO,"authenticate %s, %s from %s"%(name,password,request['REMOTE_ADDR']))

		emergency = self._emergency_user
		if name is None:
		    return None
		if emergency and name==emergency.getUserName():
		    user = emergency
		else:
		    user = self.getUser(name)
		if user is not None and user.authenticate(password, request):
		    domains = user.getDomains()
		    if self.domainSpecMatch(domains, request):
		        #zLOG.LOG('IntranetUserFolder',zLOG.INFO,"  as %s"%user)
			return user

		#zLOG.LOG('IntranetUserFolder',zLOG.INFO,"  failed!")
		return None

	def domainSpecMatch(self, spec, request):
	    """modified domainspecmatch to look at FORWARDED_FOR"""
	    #zLOG.LOG('IntranetUserFolder',zLOG.INFO,"domainspecmatch %s, %s"%(self,spec))
	    host=''
	    addr=''


	    # Fast exit for the match-all case
	    if len(spec) == 1 and spec[0] == '*':
		return 1

	    if request.has_key('REMOTE_HOST'):
		host=request['REMOTE_HOST']

	    addr=request.getClientAddr()
	    #if request.has_key('REMOTE_ADDR'):
	    #    addr=request['REMOTE_ADDR']

	    if request.has_key('HTTP_X_FORWARDED_FOR'):
		addr=request['HTTP_X_FORWARDED_FOR']
	        #zLOG.LOG('IntranetUserFolder',zLOG.INFO,"forwarded addr: %s"%(addr))
		# check for strange headers (may be fake)
		if len(addr.split('.')) != 4:
		    zLOG.LOG('IntranetUserFolder',zLOG.WARNING,"invalid forward addr: %s"%(addr))
		    return 0
	    
	    if not host and not addr:
		return 0

	    if not host:
		try:    host=socket.gethostbyaddr(addr)[0]
		except: pass
	    if not addr:
		try:    addr=socket.gethostbyname(host)
		except: pass

	    _host=host.split('.')
	    _addr=addr.split('.')
	    _hlen=len(_host)
	    _alen=len(_addr)

	    #zLOG.LOG('IntranetUserFolder',zLOG.INFO,"host: %s, addr: %s"%(_host,_addr))

	    for ob in spec:
		sz=len(ob)
		_ob=ob.split('.')
		_sz=len(_ob)

		mo = addr_match(ob)
		if mo is not None:
		    if mo.end(0)==sz:
			fail=0
			for i in range(_sz):
			    a=_addr[i]
			    o=_ob[i]
			    if (o != a) and (o != '*'):
				fail=1
				break
			if fail:
			    continue
			return 1

		mo = host_match(ob)
		if mo is not None:
		    if mo.end(0)==sz:
			if _hlen < _sz:
			    continue
			elif _hlen > _sz:
			    _item=_host[-_sz:]
			else:
			    _item=_host
			fail=0
			for i in range(_sz):
			    h=_item[i]
			    o=_ob[i]
			    if (o != h) and (o != '*'):
				fail=1
				break
			if fail:
			    continue
			return 1
	    return 0

Globals.default__class_init__(IntranetUserFolder)


def manage_addIntranetUserFolder(self,dtself=None,REQUEST=None,**ignored):
    """add a user folder """
    f=IntranetUserFolder()
    self=self.this()
    try:    self._setObject('acl_users', f)
    except: return MessageDialog(
                   title  ='Item Exists',
                   message='This object already contains a User Folder',
                   action ='%s/manage_main' % REQUEST['URL1'])
    self.__allow_groups__=f
    if REQUEST is not None:
        REQUEST['RESPONSE'].redirect(self.absolute_url()+'/manage_main')

def manage_addIntranetUserFolderForm(self):
	"""add a user folder form"""
	return manage_addIntranetUserFolder(self,REQUEST=self.REQUEST)

addr_match=re.compile(r'((\d{1,3}\.){1,3}\*)|((\d{1,3}\.){3}\d{1,3})').match
host_match=re.compile(r'(([\_0-9a-zA-Z\-]*\.)*[0-9a-zA-Z\-]*)').match