Образовательный проект «SnakeProject» Михаила Козлова

⇒ Python ⇐

CISCO

Voice(Asterisk\Cisco)

Microsoft

Powershell

SQL\T-SQL

FreeBSD and Nix

1С

Общая

WEB Разработка

ORACLE SQL \ JAVA

Мото

Стрельба, пневматика, оружие

Саморазвитие и психология

Python HTTP и HTTPS proxy server на Twisted

За основу был взят скрипт с этого проекта: https://github.com/fmoo/twisted-connect-proxy за что огромное спасибо автору!

Я его допилил под свои нужды сделав небольшой "топорный" функционал по блокированию совпадений в запросах.

Получилось подобие ACL листов, вообщем в итоге имеем HTTP\HTTPS прокси-север на python 2.7 с возможностью блокировать для определенных ip адресов определенные совпадения в запросах.

from twisted.internet.protocol import Protocol, ClientFactory from twisted.web.proxy import Proxy, ProxyRequest from twisted.internet import reactor import urlparse # from twisted.python import log # import sys # log.startLogging(sys.stdout) blacklist1 = ['youtube', 'vk.com', 'odnoklassniki', 'facebook'] ips1 = ['192.168.45.2', '192.168.45.69'] class ConnectProxyRequest(ProxyRequest): connectedProtocol = None def restrict(self, url, lst): for x in lst: if x in url: return 'y' return 'n' def detected(self, url, ip): if ip in ips1: return self.restrict(url, blacklist1) return 'n' def process(self): if self.detected( self.getAllHeaders()['host'], self.getClientIP() ) == 'y' : print "Request from %s for %s detected!" % (self.getClientIP(), self.getAllHeaders()['host']) return else: print "Request from %s for %s pass!" % (self.getClientIP(), self.getAllHeaders()['host']) if self.method == 'CONNECT': self.processConnectRequest() else: ProxyRequest.process(self) def fail(self, message, body): self.setResponseCode(501, message) self.responseHeaders.addRawHeader("Content-Type", "text/html") self.write(body) self.finish() def splitHostPort(self, hostport, default_port): port = default_port parts = hostport.split(':', 1) if len(parts) == 2: try: port = int(parts[1]) except ValueError: pass return parts[0], port def processConnectRequest(self): parsed = urlparse.urlparse(self.uri) default_port = self.ports.get(parsed.scheme) host, port = self.splitHostPort(parsed.netloc or parsed.path, default_port) if port is None: self.fail("Bad connection request", "Unable to parse port from URI: %s" % repr(self.uri)) return clientFactory = ConnectProxyClientFactory(host, port, self) self.reactor.connectTCP(host, port, clientFactory) class ConnectProxy(Proxy): requestFactory = ConnectProxyRequest connectedRemote = None def requestDone(self, request): if request.method == 'CONNECT' and self.connectedRemote is not None: self.connectedRemote.connectedClient = self else: Proxy.requestDone(self, request) def connectionLost(self, reason): if self.connectedRemote is not None: self.connectedRemote.transport.loseConnection() Proxy.connectionLost(self, reason) def dataReceived(self, data): if self.connectedRemote is None: Proxy.dataReceived(self, data) else: self.connectedRemote.transport.write(data) class ConnectProxyClient(Protocol): connectedClient = None def connectionMade(self): self.factory.request.channel.connectedRemote = self self.factory.request.setResponseCode(200, "CONNECT OK") self.factory.request.setHeader('X-Connected-IP', self.transport.realAddress[0]) self.factory.request.setHeader('Content-Length', '0') self.factory.request.finish() def connectionLost(self, reason): if self.connectedClient is not None: self.connectedClient.transport.loseConnection() def dataReceived(self, data): if self.connectedClient is not None: self.connectedClient.transport.write(data) class ConnectProxyClientFactory(ClientFactory): protocol = ConnectProxyClient def __init__(self, host, port, request): self.request = request self.host = host self.port = port def clientConnectionFailed(self, connector, reason): self.request.fail("Gateway Error", str(reason)) if __name__ == '__main__': import argparse ap = argparse.ArgumentParser() ap.add_argument('port', default=8080, nargs='?', type=int) ap.add_argument('--ssl-cert', type=str) ap.add_argument('--ssl-key', type=str) ns = ap.parse_args() import twisted.web.http factory = twisted.web.http.HTTPFactory() factory.protocol = ConnectProxy import twisted.internet if ns.ssl_cert: from twisted.internet import ssl with open(ns.ssl_cert, 'rb') as fp: ssl_cert = fp.read() if ns.ssl_key: from OpenSSL import crypto with open(ns.ssl_key, 'rb') as fp: ssl_key = fp.read() certificate = ssl.PrivateCertificate.load(ssl_cert, ssl.KeyPair.load(ssl_key, crypto.FILETYPE_PEM), crypto.FILETYPE_PEM) else: certificate = ssl.PrivateCertificate.loadPEM(ssl_cert) twisted.internet.reactor.listenSSL(ns.port, factory, certificate.options()) else: twisted.internet.reactor.listenTCP(ns.port, factory) twisted.internet.reactor.run()

Комментарии пользователей