Skip to content

Instantly share code, notes, and snippets.

@questjay
Last active October 20, 2023 02:30
Show Gist options
  • Save questjay/3f858c2fea1731d29ea20cd5cb444e30 to your computer and use it in GitHub Desktop.
Save questjay/3f858c2fea1731d29ea20cd5cb444e30 to your computer and use it in GitHub Desktop.
create proxy server on flask and requests python
"""today am gonna show you how to create a http proxy in flask, if your want https ssl proxy please dope a email at info[at]junno[dot]co"""
"""and please make sure to follow me for more proxies server and more server gists"""
""" add the proxy url to firefox,chrome or any other browser eg(http://proxy_flask_address:proxy_flask_port5000)"""
"""remember on replit this works only on locally if you want it to work publically please purchase a server e.g(linode, digitalocean etc..)"""
"""curl -x http://172.18.0.9:5000 -L http://azenv.net -vs >curl-output.txt 2>&1
"""SSL is the old name. It is called TLS these days."""
"""to use ssl add ssl_context='adhoc' to your app.run() call or $ flask run --cert=adhoc and then To use ad hoc certificates with Flask, you need to install an additional dependency in your virtual environment:
$ pip install pyopenssl"""
import re
from urllib.parse import urlparse, urlunparse
from flask import Flask, render_template, request, abort, Response, redirect
import requests
import logging
app = Flask('')
@app.route('/', methods=["GET", "POST"])
def root():
# If referred from a proxy request, then redirect to a URL with the proxy prefix.
# This allows server-relative and protocol-relative URLs to work.
referer = request.headers.get('referer')
if not request.url:
return Response("Relative URL sent without a a proxying request referal. Please specify a valid proxy host (/p/url)", 400)
parts = None
if request.url == "":
parts = referer
else:
parts = request.url
#return redirect(urlunparse(parts._replace(path=parts.path+'/')))
r = make_request(part, request.method, dict(request.headers), request.form)
headers = dict(r.raw.headers)
h = filter_headers(headers)
def generate():
for chunk in r.raw.stream(decode_content=False):
yield chunk
out = app.response_class(generate(), headers=headers)
print(out)
out.status_code = r.status_code
return out #(r.text, r.status_code, headers)
def filter_headers(headers):
# http://tools.ietf.org/html/rfc2616#section-13.5.1
hop_by_hop = ('Connection', 'Keep-Alive','Te', 'Trailers', 'Transfer-Encoding', 'Upgrade')
for k in hop_by_hop:
if k in headers:
del headers[k]
# accept only supported encodings
if 'Accept-Encoding' in headers:
ae = headers['Accept-Encoding']
filtered_encodings = [x for x in re.split(r',\s*', ae) if x in ('identity', 'gzip', 'x-gzip', 'deflate')]
headers['Accept-Encoding'] = ', '.join(filtered_encodings)
return headers
def make_request(url, method, headers={}, data=None):
# Ensure the URL is approved, else abort
# Pass original Referer for subsequent resource requests
try:
referer = request.headers.get('referer')
if referer:
headers.update({ "referer" : url})
# Fetch the URL, and stream it back
LOG.debug("Sending %s %s with headers: %s and data %s", method, url, headers, data)
return requests.request(method, url, params=request.args, stream=True, headers=headers, allow_redirects=False, data=data)
except Exception as e:
print(e)
def run():
print('starting')
app.run(host = '0.0.0.0', port = 5000)
def keep_alive():
from threading import Thread
t = Thread(target = run)
t.start()
if __name__ == '__main__':
run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment