Python werkzeug.contrib.fixers 模块，ProxyFix() 实例源码

def init_app(cls, app):
        ProductionConfig.init_app(app)

        # handle proxy server headers
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

        ## log to syslog
        # write to /var/log/messages
        # can be configured to write to a separate log file
        # see docs
        import logging
        from logging.handlers import SysLogHandler
        syslog_handler = SysLogHandler()
        syslog_handler.setLevel(logging.WARNING)
        app.logger.addHandler(syslog_handler)

def run(self):
        bind_addr = (self.config['listen'], self.config['port'])

        wsgi_app = ReverseProxied(ProxyFix(wsgi.WSGIPathInfoDispatcher({'/': app})))
        self.server = wsgi.WSGIServer(bind_addr=bind_addr,
                                      wsgi_app=wsgi_app)
        self.server.ssl_adapter = http_helpers.ssl_adapter(self.config['certificate'],
                                                           self.config['private_key'])
        logger.debug('WSGIServer starting... uid: %s, listen: %s:%s', os.getuid(), bind_addr[0], bind_addr[1])
        for route in http_helpers.list_routes(app):
            logger.debug(route)

        try:
            self.server.start()
        except KeyboardInterrupt:
            self.server.stop()

def load(self, proxy_mode=False, rules=None, newrelic_agent=None):
        _logger.info("Loading Odoo WSGI application")
        self.load_server_wide_modules()

        application = odoo_application

        if config['debug_mode']:
            application = DebuggedApplication(application, evalex=True)
            _logger.warning("Debugger enabled, do not use in production")

        if newrelic_agent:
            application = newrelic_agent.WSGIApplicationWrapper(application)
            _logger.info("New Relic enabled")

        if rules and rules.has_rules():
            application = rules(application)
            _logger.info("Rewrites enabled")

        if proxy_mode:
            application = ProxyFix(application)
            _logger.info("Proxy mode enabled")

        return application

def init_app(cls, app):
        Config.init_app(app)

        # ?????????
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

def init_app(cls, app):
        ProductionConfig.init_app(app)

        # handle proxy server headers
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

        # log to stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.WARNING)
        app.logger.addHandler(file_handler)

def setup_captive_portal_app():
    cpm = Flask(__name__)
    cpm.add_url_rule('/success.html',
                     'success',
                     cp_check_ios_lt_v9_macos_lt_v1010)
    # iOS from captive portal
    cpm.add_url_rule('/library/test/success.html',
                     'success',
                     cp_check_ios_lt_v9_macos_lt_v1010)
    cpm.add_url_rule('/hotspot-detect.html',
                     'hotspot-detect', cp_check_ios_gte_v9_macos_gte_v1010)
    # Android <= v6 (possibly later too)
    # noqa: See: https://www.chromium.org/chromium-os/chromiumos-design-docs/network-portal-detection
    cpm.add_url_rule('/generate_204', 'welcome',
                     show_captive_portal_welcome)
    # Fallback method introduced in Android 7
    # See:
    # noqa: https://android.googlesource.com/platform/frameworks/base/+/master/services/core/java/com/android/server/connectivity/NetworkMonitor.java#92
    cpm.add_url_rule('/gen_204', 'welcome',
                     show_captive_portal_welcome)
    # Captive Portal check for Amazon Kindle Fire
    cpm.add_url_rule('/kindle-wifi/wifistub.html', 'welcome',
                     show_captive_portal_welcome)
    # Captive Portal check for Windows
    # See: https://technet.microsoft.com/en-us/library/cc766017(v=ws.10).aspx
    cpm.add_url_rule('/ncsi.txt', 'welcome',
                     show_captive_portal_welcome)
    # cpm.add_url_rule('/_authorised_clients',
    #                  'auth', get_authorised_clients, methods=['GET'])
    cpm.add_url_rule('/_authorised_clients',
                     'auth', add_authorised_client, methods=['POST'])
    cpm.add_url_rule('/_authorised_clients/<ip_addr_str>',
                     'auth_ip', add_authorised_client, methods=['PUT'])
    cpm.add_url_rule('/_authorised_clients',
                     'deauth', remove_authorised_client, methods=['DELETE'])
    cpm.add_url_rule('/_authorised_clients/<ip_addr_str>',
                     'deauth_ip', remove_authorised_client, methods=['DELETE'])
    cpm.add_url_rule('/_redirect_to_connectbox',
                     'redirect', redirect_to_connectbox)
    cpm.wsgi_app = ProxyFix(cpm.wsgi_app)
    return cpm

def create_app():
    logging.info('OS-FDP-ADAPTERS create_app')
    app = Flask('os_fdp_adapters')
    app.wsgi_app = ProxyFix(app.wsgi_app)
    logging.info('OS-API configuring blueprints')
    app.register_blueprint(OSFdpAdapter, url_prefix='/')
    CORS(app)
    logging.info('OS-FDP-ADAPTERS app created')
    return app

def init_app(cls, app):
        ProductionConfig.init_app(app)

        # ?????????
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)
        # ???stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.WARNING)
        app.logger.addHandler(file_handler)

def init_app(cls, app):
        ProductionConfig.init_app(app)

        # handle proxy server headers
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

        # log to stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.WARNING)
        app.logger.addHandler(file_handler)

def test_proxy_fix(self):
        @Request.application
        def app(request):
            return Response('%s|%s' % (
                request.remote_addr,
                # do not use request.host as this fixes too :)
                request.environ['HTTP_HOST']
            ))
        app = fixers.ProxyFix(app, num_proxies=2)
        environ = dict(create_environ(),
            HTTP_X_FORWARDED_PROTO="https",
            HTTP_X_FORWARDED_HOST='example.com',
            HTTP_X_FORWARDED_FOR='1.2.3.4, 5.6.7.8',
            REMOTE_ADDR='127.0.0.1',
            HTTP_HOST='fake'
        )

        response = Response.from_app(app, environ)

        self.assert_equal(response.get_data(), b'1.2.3.4|example.com')

        # And we must check that if it is a redirection it is
        # correctly done:

        redirect_app = redirect('/foo/bar.hml')
        response = Response.from_app(redirect_app, environ)

        wsgi_headers = response.get_wsgi_headers(environ)
        assert wsgi_headers['Location'] == 'https://example.com/foo/bar.hml'

def test_proxy_fix_weird_enum(self):
        @fixers.ProxyFix
        @Request.application
        def app(request):
            return Response(request.remote_addr)
        environ = dict(create_environ(),
            HTTP_X_FORWARDED_FOR=',',
            REMOTE_ADDR='127.0.0.1',
        )

        response = Response.from_app(app, environ)
        self.assert_strict_equal(response.get_data(), b'127.0.0.1')

def _create_app():
    """
    Flask Application factory.
    :return: Flask app for an API.
    :rtype: flask.Flask
    """
    app = flask.Flask(__name__)
    # http://werkzeug.pocoo.org/docs/0.12/contrib/fixers/#werkzeug.contrib.fixers.ProxyFix
    app.wsgi_app = ProxyFix(app.wsgi_app)
    return app

def init_app(cls, app):
        ProductionConfig.init_app(app)

        # handle proxy server headers
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

        # log to stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.WARNING)
        app.logger.addHandler(file_handler)

def init_app(cls, app):
        ProductionConfig.init_app(app)
        SSL_DISABLE = bool(os.environ.get('SSL_DISABLE'))
        # log to stderr
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.WARNING)
        app.logger.addHandler(file_handler)

def init_app(cls, app):
        ProductionConfig.init_app(app)

        # handle reverse proxy server headers
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

        # log to stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)

def run(self):
        wsgi_app_https = ReverseProxied(ProxyFix(wsgi.WSGIPathInfoDispatcher({'/': app})))
        wsgi_app_http = ReverseProxied(ProxyFix(wsgi.WSGIPathInfoDispatcher({'/': adapter_app})))
        cherrypy.server.unsubscribe()
        cherrypy.config.update({'environment': 'production'})

        bind_addr = (self.config['listen'], self.config['port'])

        server_https = wsgi.WSGIServer(bind_addr=bind_addr,
                                       wsgi_app=wsgi_app_https)
        server_https.ssl_adapter = http_helpers.ssl_adapter(self.config['certificate'],
                                                            self.config['private_key'])
        ServerAdapter(cherrypy.engine, server_https).subscribe()
        logger.debug('WSGIServer starting... uid: %s, listen: %s:%s',
                     os.getuid(), bind_addr[0], bind_addr[1])

        for route in http_helpers.list_routes(app):
            logger.debug(route)

        if self.adapter_config['enabled']:
            bind_addr = (self.adapter_config['listen'], self.adapter_config['port'])
            server_adapter = wsgi.WSGIServer(bind_addr=bind_addr,
                                             wsgi_app=wsgi_app_http)
            ServerAdapter(cherrypy.engine, server_adapter).subscribe()
            logger.debug('WSGIServer starting... uid: %s, listen: %s:%s',
                         os.getuid(), bind_addr[0], bind_addr[1])

            for route in http_helpers.list_routes(adapter_app):
                logger.debug(route)

        else:
            logger.debug('Adapter server is disabled')

        try:
            cherrypy.engine.start()
            cherrypy.engine.wait(states.EXITING)
        except KeyboardInterrupt:
            logger.warning('Stopping xivo-ctid-ng: KeyboardInterrupt')
            cherrypy.engine.exit()

def init_app(cls, app):
        ProductionConfig.init_app(app)

        # Handle proxy server headers
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

def create_app():
    _app = Flask(__name__)

    # used for encrypting cookies for handling sessions
    _app.config['SECRET_KEY'] = 'abc492ee-9739-11e6-a174-07f6b92d4a4b'

    message_queue_type = environ.env.config.get(ConfigKeys.TYPE, domain=ConfigKeys.QUEUE, default=None)
    if message_queue_type is None and not (len(environ.env.config) == 0 or environ.env.config.get(ConfigKeys.TESTING)):
        raise RuntimeError('no message queue type specified')

    message_queue = 'redis://%s' % environ.env.config.get(ConfigKeys.HOST, domain=ConfigKeys.CACHE_SERVICE, default='')
    message_channel = 'dino_%s' % environ.env.config.get(ConfigKeys.ENVIRONMENT, default='test')

    logger.info('message_queue: %s' % message_queue)

    _api = Api(_app)

    _socketio = SocketIO(
            _app,
            logger=logger,
            engineio_logger=os.environ.get('DINO_DEBUG', '0') == '1',
            async_mode='eventlet',
            message_queue=message_queue,
            channel=message_channel)

    # preferably "emit" should be set during env creation, but the socketio object is not created until after env is
    environ.env.out_of_scope_emit = _socketio.emit

    _app.wsgi_app = ProxyFix(_app.wsgi_app)
    return _app, _api, _socketio

def create_app():
    _app = Flask(__name__)

    # used for encrypting cookies for handling sessions
    _app.config['SECRET_KEY'] = 'abc492ee-9739-11e6-a174-07f6b92d4a4b'

    message_queue_type = environ.env.config.get(ConfigKeys.TYPE, domain=ConfigKeys.QUEUE, default=None)
    if message_queue_type is None and not (len(environ.env.config) == 0 or environ.env.config.get(ConfigKeys.TESTING)):
        raise RuntimeError('no message queue type specified')

    message_queue = 'redis://%s' % environ.env.config.get(ConfigKeys.HOST, domain=ConfigKeys.CACHE_SERVICE, default='')
    message_channel = 'dino_%s' % environ.env.config.get(ConfigKeys.ENVIRONMENT, default='test')

    logger.info('message_queue: %s' % message_queue)

    _socketio = SocketIO(
            _app,
            logger=logger,
            engineio_logger=os.environ.get('DINO_DEBUG', '0') == '1',
            async_mode='eventlet',
            message_queue=message_queue,
            channel=message_channel)

    # preferably "emit" should be set during env creation, but the socketio object is not created until after env is
    environ.env.out_of_scope_emit = _socketio.emit

    _app.wsgi_app = ProxyFix(_app.wsgi_app)
    return _app, _socketio

def create_app():
    _app = Flask(
            import_name=__name__,
            template_folder='admin/templates/',
            static_folder='admin/static/')

    # used for encrypting cookies for handling sessions
    _app.config['SECRET_KEY'] = 'abc492ee-9739-11e6-a174-07f6b92d4a4b'
    _app.config['ROOT_URL'] = environ.env.config.get(ConfigKeys.ROOT_URL, domain=ConfigKeys.WEB, default='/')

    message_queue_type = environ.env.config.get(ConfigKeys.TYPE, domain=ConfigKeys.QUEUE, default=None)
    if message_queue_type is None and not (len(environ.env.config) == 0 or environ.env.config.get(ConfigKeys.TESTING)):
        raise RuntimeError('no message queue type specified')

    message_queue = 'redis://%s' % environ.env.config.get(ConfigKeys.HOST, domain=ConfigKeys.CACHE_SERVICE, default='')
    message_channel = 'dino_%s' % environ.env.config.get(ConfigKeys.ENVIRONMENT, default='test')

    logger.info('message_queue: %s' % message_queue)

    _socketio = SocketIO(
            _app,
            logger=logger,
            engineio_logger=os.environ.get('DINO_DEBUG', '0') == '1',
            async_mode='eventlet',
            message_queue=message_queue,
            channel=message_channel)

    # preferably "emit" should be set during env creation, but the socketio object is not created until after env is
    environ.env.out_of_scope_emit = _socketio.emit

    _app.wsgi_app = ReverseProxied(ProxyFix(_app.wsgi_app))
    return _app, _socketio

def init_app(cls, app):
        ProductionConfig.init_app(app)

        # handle proxy server headers
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

        # log to stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.WARNING)
        app.logger.addHandler(file_handler)

def create_app(config_name=None):
    logger.info('Creating flask app...')
    app = Flask(__name__)

    if config_name is None:
        logger.info('Config name not supplied, searching environment')
        config_name = os.environ.get('SDV_APP_SETTINGS', 'development')
        logger.info('Config name set to: {}'.format(config_name))

    logger.info('Initialising extensions')
    app.config.from_object(config[config_name])
    recaptcha.init_app(app=app)
    bcrypt.init_app(app)
    mail.init_app(app)
    censor.init_app(app=app)

    app.secret_key = app.config['SECRET_KEY']
    app.jinja_env.trim_blocks = True
    app.jinja_env.lstrip_blocks = True
    app.wsgi_app = ProxyFix(app.wsgi_app)

    if app.config['USE_SQLITE']:
        logger.info('Application set to use SQLite')
        app.database = app.config['DB_SQLITE']
        app.sqlesc = '?'

        def connect_db():
            return sqlite3.connect(app.database)
    else:
        logger.info('Application set to use Postgres')
        app.database = 'dbname=' + app.config['DB_NAME'] + ' user=' + app.config[
            'DB_USER'] + ' password=' + app.config['DB_PASSWORD']
        app.sqlesc = '%s'

    return app

def init_app(cls, app):
        ProductionConfig.init_app(app)

        # Handle proxy server headers
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

    # mail.init_app(app)
    init_logging(app.config['APP_LOG_DIR'])
    app.wsgi_app = ProxyFix(app.wsgi_app)
    celery.conf.update(app.config)

    # routes and errorhandlers 
    from .main import main as main_blueprint
    app.register_blueprint(main_blueprint)

    return app

def init_app(cls, app):
        ProductionConfig.init_app(app)

        # handle proxy server headers
        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

        # log to stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.WARNING)
        app.logger.addHandler(file_handler)

# Unix?????

def init_app(cls, app):
        ProductionConfig.init_app(app)

        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.WARNING)
        app.logger.addHandler(file_handler)

        from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)

def _apply_patches(app):
    """
    Apply special fixes and/or monkey patches. These fixes will hopefully
    become obsolete with proper fixes made to these libraries in the future.
    """
    # "X-Forwarded-For" remote_ip fixup when running behind a load balancer.
    # Assuming we are running behind two proxies (or load balancers), the API
    # router, and the auto-scaling group.
    num_proxies = 1
    import socket
    if socket.gethostname().startswith("ip-10-60-"):
        num_proxies = 2
    app.wsgi_app = ProxyFix(app.wsgi_app, num_proxies=num_proxies)

    # Fixing SCRIPT_NAME/url_scheme when behind reverse proxy (i.e. the
    # API router).
    app.wsgi_app = ReverseProxied(app.wsgi_app)

    # Datetime fix
    app.json_encoder = CustomJSONEncoder

    # Euthanize Flask Restful exception handlers. This may get fixed
    # soon in "Error handling re-worked #544"
    # https://github.com/flask-restful/flask-restful/pull/544
    def patched_init_app(self, app):
        if len(self.resources) > 0:
            for resource, urls, kwargs in self.resources:
                self._register_view(app, resource, *urls, **kwargs)
    Api._init_app = patched_init_app

    # Install proper json dumper for Flask Restful library.
    # This is needed because we need to use Flask's JSON converter which can
    # handle more variety of Python types than the standard converter.
    def output_json(obj, code, headers=None):
        resp = make_response(dumps(obj, indent=4), code)
        resp.headers.extend(headers or {})
        return resp
    if isinstance(flask_restful.DEFAULT_REPRESENTATIONS, dict):
        flask_restful.DEFAULT_REPRESENTATIONS['application/json'] = output_json
    else:
        flask_restful.DEFAULT_REPRESENTATIONS = [('application/json', output_json)]

    def tenant_not_found(message):
        status_code = httplib.NOT_FOUND
        response = jsonify({"error": message, "status_code": status_code})
        response.status_code = status_code
        return response

    @app.errorhandler(TenantNotFoundError)
    def bad_request_handler(error):
        return tenant_not_found(error.message)

def create_web_app(configuration: UchanConfiguration, app):
    if configuration.http.use_proxy_fixer:
        app.wsgi_app = ProxyFix(app.wsgi_app, configuration.http.proxy_fixer_num_proxies)

    app.config['DEBUG'] = configuration.app.debug
    app.config['APP_NAME'] = configuration.app.name
    app.config['MAX_CONTENT_LENGTH'] = configuration.http.max_content_length

    app.jinja_env.trim_blocks = True
    app.jinja_env.lstrip_blocks = True

    import uchan.view.routing.converters
    uchan.view.routing.converters.init_converters(app)

    from uchan import logger
    from uchan.lib.exceptions import BadRequestError

    # Setup error handlers
    @app.errorhandler(500)
    def server_error_handler(error):
        logger.exception(error)
        return app.send_static_file('500.html'), 500

    @app.errorhandler(404)
    def server_error_handler(error):
        return app.send_static_file('404.html'), 404

    def bad_request_message(e):
        if isinstance(e, BadRequestError):
            while isinstance(e, Exception) and len(e.args) > 0:
                e = e.args[0]

        return e if type(e) is str else ''

    from uchan.view import render_error

    @app.errorhandler(BadRequestError)
    def bad_request_handler(error):
        user_message = bad_request_message(error)

        return render_error(user_message, 400)

    from uchan.lib.action_authorizer import NoPermissionError

    @app.errorhandler(NoPermissionError)
    def no_permission_handler(error):
        return render_error('No permission', 401)

    from uchan.lib.service import verification_service

    @app.after_request
    def after_request_handler(response):
        verification_service.after_request(response)
        return response

    return app