import json import urllib from nevow import tags as T, flat from twisted.internet.defer import succeed from twisted.python import log from twisted.web.client import getPage API_SERVER = '' API_SSL_SERVER = '' def _encode_if_necessary(s): if isinstance(s, unicode): return s.encode('utf-8') return s class RecaptchaResponse(object): """ Response from RecaptchaAccount.verify. Check response.is_valid. If it isn't true use response.display to display the error on the new page. """ def __init__(self, account, error): self.is_valid = error is None self.account = account self._error = error def display(self, use_nevow=False, use_ssl=False, options={}): if self._error is None: raise ValueError, "trying to redisplay a passed captcha" return self.account._display(use_nevow, use_ssl, options, self._error) class RecaptchaAccount(object): def __init__(self, public_key, private_key): self.public_key = public_key self.private_key = private_key def _display(self, use_nevow, use_ssl, options, error): server = API_SSL_SERVER if use_ssl else API_SERVER params = {'k': _encode_if_necessary(self.public_key)} if error is not None: params['error'] = error params = urllib.urlencode(params) tree = [ T.script(type='text/javascript')[ 'var RecaptchaOptions = ', json.dumps(options), ';', ], T.script(type='text/javascript', src=server + '/challenge?' + params), T.noscript[ T.iframe(src=server + '/noscript?' + params, height='300', width='500', frameborder='0'),, T.textarea(name='recaptcha_challenge_field', rows='3', cols='40'), T.input(type='hidden', name='recaptcha_response_field', value='manual_challenge'), ] ] if use_nevow: return tree return str(flat.flatten(tree)) def display(self, use_nevow=False, use_ssl=False, options={}): """ Returns the HTML to display. use_nevow -- return a nevow stan tree instead of a string use_ssl -- make the client request the captcha through ssl, do this if the page is served over ssl to avoid a warning options -- a dictionary of options, see """ return self._display(use_nevow, use_ssl, options, None) def verify(self, remoteip, challenge, response): """ Submits a request for verification. Returns a deferred which is called back with a RecaptchaResponse instance. remoteip -- the client's ip address challenge -- the value of recaptcha_challenge_field from the form response -- the value of recaptcha_response_field from the form """ if not (response and challenge and len(response) and len(challenge)): return succeed(RecaptchaResponse(account=self, error='incorrect-captcha-sol')) return getPage( url=API_SERVER + '/verify', method='POST', postdata=urllib.urlencode({ 'privatekey': _encode_if_necessary(self.private_key), 'remoteip': _encode_if_necessary(remoteip), 'challenge': _encode_if_necessary(challenge), 'response': _encode_if_necessary(response), }), headers={'Content-Type': 'application/x-www-form-urlencoded'}, agent='reCAPTCHA Twisted Python', ).addCallback(self._handle_response) def _handle_response(self, response): return_values = response.splitlines() if return_values[0] == 'true': return RecaptchaResponse(self, None) if return_values[0] == 'false': log.err(ValueError("Recaptcha error: " + repr(return_values[1]))) # because this can indicate the keys are incorrect return RecaptchaResponse(self, return_values[1]) raise ValueError("invalid response: %r" % (response,)) if __name__ == '__main__': from nevow import rend, tags as T, inevow, loaders, url from nevow.appserver import NevowSite from twisted.internet import reactor from zope.interface import implements captcha_account = RecaptchaAccount('6LeaFLsSAAAAAA9BlCwD_q5lqBuyskWdAg-dQOPJ', '6LeaFLsSAAAAADvMV7iYPGSuaw6iTYRDS_7GvFD9') class Passed(rend.Page): docFactory = loaders.stan( T.html[ T.body[ T.h1['passed!'], T.p[T.a(href='')['home']], ] ] ) class Test(rend.Page): docFactory = loaders.xmlstr("""
""") def __init__(self, captcha_source): rend.Page.__init__(self) self.captcha_source = captcha_source def render_captcha(self, ctx, data): return self.captcha_source.display(use_nevow=True, options={'theme': 'white'}) class Root(object): implements(inevow.IResource) def locateChild(self, ctx, segments): request = inevow.IRequest(ctx) if request.method == "GET": return Test(captcha_account), () return captcha_account.verify( request.getClientIP(), request.args['recaptcha_challenge_field'][0], request.args['recaptcha_response_field'][0], ).addCallback(self.handle_response) def handle_response(self, response): if response.is_valid: return Passed(), () return Test(response), () reactor.listenTCP(8080, NevowSite(Root()))