<commit_before>from __future__ import absolute_import
import logging
import sys
import os
import socket
import time
import select
import threading
from hashlib import md5, sha1
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer, SimpleHandler, make_server

import kaa
from . import bottle
from .bottle import route

# Bottle assumes it's in sys.path for error pages.  Add it to sys.modules so
# import inside the bottle error template can find it.
sys.modules['bottle'] = bottle

log = logging.getLogger('stagehand.web')

class LoggingMiddleware(object):
    def __init__(self, application, logger):
        self._application = application
        if isinstance(logger, logging.Logger):
            self._httplog = logger
        else:
            handler = logging.StreamHandler()
            handler.setFormatter(logging.Formatter('%(asctime)s [HTTP] %(message)s'))
            self._httplog = logging.getLogger(logger)
            self._httplog.addHandler(handler)
            self._httplog.propagate = False


    def __call__(self, environ, start_response):
        # Default log level for this call.  The handler can modify it.
        bottle.response.loglevel = logging.INFO
        bottle.response.logextra = None
        t0 = time.time()

        def _start_response(status, headers, *args):
            uri = environ['PATH_INFO'] + ('?%s' % environ['QUERY_STRING'] if environ['QUERY_STRING'] else '')
            code = status.split()[0]
            size = dict(headers).get('Content-Length', '-')

            self._httplog.log(bottle.response.loglevel, '%s "%s %s %s" %s %s %.03fs%s', 
                              environ['REMOTE_ADDR'], environ['REQUEST_METHOD'],
                              uri, environ['SERVER_PROTOCOL'], code, size, time.time()-t0,
                              ' ' + bottle.response.logextra if bottle.response.logextra else '')
            return start_response(status, headers, *args)

        try:
            return self._application(environ, _start_response)
        except Exception, r:
            log.exception('uncaught WSGI exception')



class AuthMiddleware(object):
    def __init__(self, application, user, passwd):
        self._application = application
        self._nonce_passwd = os.urandom(64)
        self._user = user
        self._passwd = <PASSWORD>


    def __call__(self, environ, start_response):
        try:
            nexthop = self._application if self._check_auth(environ) else self._auth_failed
            return nexthop(environ, start_response)
        except Exception, e:
            log.exception('error handling authentication')
            return []


    def _check_auth(self, environ):
        response = environ.get('HTTP_AUTHORIZATION')
        if not response or not response.lower().startswith('digest '):
            return False
        # Parse response into a list of 2-tuples (key, value)
        parts = (f.strip().split('=', 1) for f in response[7:].split(',') if '=' in f)
        # Remove enclosing quotes and toss into a dict.
        rdict = dict((k.lower(), v.strip('"')) for k, v in parts)
        HA1 = md5('%s:%s:%s' % (self._user, rdict.get('realm'), self._passwd)).hexdigest()
        HA2 = md5('%s:%s' % (environ.get('REQUEST_METHOD'), rdict.get('uri'))).hexdigest()
        expected = md5('%s:%s:%s' % (HA1, rdict.get('nonce'), HA2)).hexdigest()
        return rdict.get('response') == expected


    def _auth_failed(self, environ, start_response):
        now = str(time.time())
        nonce = '%s/%s' % (now, md5(str(now) + self._nonce_passwd).hexdigest())
        # TODO: advertise qop and support cnonce
        response_headers = [
            ('WWW-Authenticate', 'Digest realm="Secure Area", nonce="%s", algorithm=MD5' % nonce),
            ('Content-Type', 'text/html')
        ]
        start_response('401 Authorization Required', response_headers)
        return ["<html><body>Authorization Required</body></html>"]



class UserDataMiddleware(object):
    def __init__(self, application, userdict):
        self._application = application
        self._userdict = userdict


    def __call__(self, environ, start_response):
        environ.update(self._userdict)
        return self._application(environ, start_response)


class NonBlockingWSGIServer(WSGIServer):
    """
    Implements a non-blocking variant of the batteries-included simple WSGI
    server.  This one supports HTTP keep-alive and pipelining, as well as
    sendfile() so that transfers of large files are feasible.

    However it doesn't support POSTing of binary data yet.
    """
    def __init__(self, *args, **kwargs):
        WSGIServer.__init__(self, *args, **kwargs)

        # NonBlockingClient -> timestamp
        self.clients = {}
        # Number of simultaneous clients.
        self.limit = 1000
        # Number of seconds before we bounce idle clients.
        self.timeout = 60
        self._shutdown_event = threading.Event()


    def shutdown(self):
        if self._running:
            self._running = False
            # Poke the socket to wake up the socket server loop.  FIXME: IPv6
            socket.socket().connect(self.socket.getsockname())
            self._shutdown_event.wait()


    def serve_forever(self, poll_interval=30):
        host, port = self.server_address
        self._running = True
        self._shutdown_event.clear()
        while self._running:
            self.kill_idle_clients()
            rfds = [self] + self.clients.keys()
            wfds = [client for client in self.clients if client.writes_pending]
            # TODO: epoll instead
            r, w, e = select.select(rfds, wfds, [], poll_interval)
            for fd in r:
                if fd == self:
                    self.get_request()
                else:
                    try:
                        fd.handle_read()
                        if fd.requests_pending > 1:
                            log.debug2('pipelined [%d] %d', fd.fileno(), fd.requests_pending)
                            print '******** pipelined [%d] %d' % (fd.fileno(), fd.requests_pending)
                        # Process all fully buffered requests (if more than 1,
                        # client is using pipelining)
                        while fd.requests_pending:
                            self._handle_request(fd)
                    except:
                        log.exception('unhandled error during client request')
                        self.handle_error(client, client.addr)
                        fd.close()
            
            for fd in w:<commit_msg>Disconnect all clients on server shutdown<commit_after>from __future__ import absolute_import
import logging
import sys
import os
import socket
import time
import select
import threading
from hashlib import md5, sha1
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer, SimpleHandler, make_server

import kaa
from . import bottle
from .bottle import route

# Bottle assumes it's in sys.path for error pages.  Add it to sys.modules so
# import inside the bottle error template can find it.
sys.modules['bottle'] = bottle

log = logging.getLogger('stagehand.web')

class LoggingMiddleware(object):
    def __init__(self, application, logger):
        self._application = application
        if isinstance(logger, logging.Logger):
            self._httplog = logger
        else:
            handler = logging.StreamHandler()
            handler.setFormatter(logging.Formatter('%(asctime)s [HTTP] %(message)s'))
            self._httplog = logging.getLogger(logger)
            self._httplog.addHandler(handler)
            self._httplog.propagate = False


    def __call__(self, environ, start_response):
        # Default log level for this call.  The handler can modify it.
        bottle.response.loglevel = logging.INFO
        bottle.response.logextra = None
        t0 = time.time()

        def _start_response(status, headers, *args):
            uri = environ['PATH_INFO'] + ('?%s' % environ['QUERY_STRING'] if environ['QUERY_STRING'] else '')
            code = status.split()[0]
            size = dict(headers).get('Content-Length', '-')

            self._httplog.log(bottle.response.loglevel, '%s "%s %s %s" %s %s %.03fs%s', 
                              environ['REMOTE_ADDR'], environ['REQUEST_METHOD'],
                              uri, environ['SERVER_PROTOCOL'], code, size, time.time()-t0,
                              ' ' + bottle.response.logextra if bottle.response.logextra else '')
            return start_response(status, headers, *args)

        try:
            return self._application(environ, _start_response)
        except Exception, r:
            log.exception('uncaught WSGI exception')



class AuthMiddleware(object):
    def __init__(self, application, user, passwd):
        self._application = application
        self._nonce_passwd = <PASSWORD>(64)
        self._user = user
        self._passwd = <PASSWORD>


    def __call__(self, environ, start_response):
        try:
            nexthop = self._application if self._check_auth(environ) else self._auth_failed
            return nexthop(environ, start_response)
        except Exception, e:
            log.exception('error handling authentication')
            return []


    def _check_auth(self, environ):
        response = environ.get('HTTP_AUTHORIZATION')
        if not response or not response.lower().startswith('digest '):
            return False
        # Parse response into a list of 2-tuples (key, value)
        parts = (f.strip().split('=', 1) for f in response[7:].split(',') if '=' in f)
        # Remove enclosing quotes and toss into a dict.
        rdict = dict((k.lower(), v.strip('"')) for k, v in parts)
        HA1 = md5('%s:%s:%s' % (self._user, rdict.get('realm'), self._passwd)).hexdigest()
        HA2 = md5('%s:%s' % (environ.get('REQUEST_METHOD'), rdict.get('uri'))).hexdigest()
        expected = md5('%s:%s:%s' % (HA1, rdict.get('nonce'), HA2)).hexdigest()
        return rdict.get('response') == expected


    def _auth_failed(self, environ, start_response):
        now = str(time.time())
        nonce = '%s/%s' % (now, md5(str(now) + self._nonce_passwd).hexdigest())
        # TODO: advertise qop and support cnonce
        response_headers = [
            ('WWW-Authenticate', 'Digest realm="Secure Area", nonce="%s", algorithm=MD5' % nonce),
            ('Content-Type', 'text/html')
        ]
        start_response('401 Authorization Required', response_headers)
        return ["<html><body>Authorization Required</body></html>"]



class UserDataMiddleware(object):
    def __init__(self, application, userdict):
        self._application = application
        self._userdict = userdict


    def __call__(self, environ, start_response):
        environ.update(self._userdict)
        return self._application(environ, start_response)


class NonBlockingWSGIServer(WSGIServer):
    """
    Implements a non-blocking variant of the batteries-included simple WSGI
    server.  This one supports HTTP keep-alive and pipelining, as well as
    sendfile() so that transfers of large files are feasible.

    However it doesn't support POSTing of binary data yet.
    """
    def __init__(self, *args, **kwargs):
        WSGIServer.__init__(self, *args, **kwargs)

        # NonBlockingClient -> timestamp
        self.clients = {}
        # Number of simultaneous clients.
        self.limit = 1000
        # Number of seconds before we bounce idle clients.
        self.timeout = 60
        self._shutdown_event = threading.Event()


    def shutdown(self):
        if not self._running:
            return
        self._running = False
        # Dump all clients.
        for client in self.clients.keys():
            # Calling close() on client will remove client from self.clients
            client.close(force=True)
        # Poke the socket to wake up the socket server loop.  FIXME: IPv6
        socket.socket().connect(self.socket.getsockname())
        self._shutdown_event.wait()


    def serve_forever(self, poll_interval=30):
        host, port = self.server_address
        self._running = True
        self._shutdown_event.clear()
        while self._running:
            self.kill_idle_clients()
            rfds = [self] + self.clients.keys()
            wfds = [client for client in self.clients if client.writes_pending]
            # TODO: epoll instead
            r, w, e = select.select(rfds, wfds, [], poll_interval)
            for fd in r:
                if fd == self:
                    self.get_request()
                else:
                    try:
                        fd.handle_read()
                        if fd.requests_pending > 1:
                            log.debug2('pipelined [%d] %d', fd.fileno(), fd.requests_pending)
                            print '******** pipelined [%d] %d' % (fd.fileno(), fd.requests_pending)
                        # Process all fully buffered requests (if more than 1,
                        # client is using pipelining)
                        while fd.requests_pending:
                            self._handle_request(fd)
                    except:
                        log.exception('unhandled error during client request')
                        self.handle_error(client, client.addr)
                        fd.close()
            
            for fd in w:
{'fraction_non_alphanumeric': 0.10858849887976102, 'fraction_numerical': 0.005377147124719941, 'mean_word_length': 2.208962377186676, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5487861', 'n_tokens_mistral': 3798, 'n_tokens_neox': 3668, 'n_words': 993}