Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-23 08:11:00 +00:00
Code Issues Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master'

Browse Source 
Conflicts:
	setup.py
This commit is contained in:
Maximilian Hils 2014-11-11 12:29:34 +01:00
commit 8635e00175
19 changed files with 1262 additions and 864 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
CHANGELOG
View File

@ -1,3 +1,12 @@
7 November 2014: pathod 0.11:
* Hugely improved SSL support, including dynamic generation of certificates
using the mitproxy cacert
* pathoc -S dumps information on the remote SSL certificate chain
* Big improvements to fuzzing, including random spec selection and memoization to avoid repeating randomly generated patterns
* Reflected patterns, allowing you to embed a pathod server response specification in a pathoc request, resolving both on client side. This makes fuzzing proxies and other intermediate systems much better.
25 August 2013: pathod 0.9.2:
* Adapt to interface changes in netlib

23
libpathod/app.py
View File

@ -4,6 +4,8 @@ import version, language, utils
from netlib import http_uastrings
logging.basicConfig(level="DEBUG")
def make_app(noapi):
app = Flask(__name__)
@ -14,20 +16,17 @@ def make_app(noapi):
version = version.IVERSION
)
@app.route('/api/log')
def api_log():
return jsonify(
log = app.config["pathod"].get_log()
)
@app.route('/api/clear_log')
def api_clear_log():
app.config["pathod"].clear_log()
return "OK"
def render(s, cacheable, **kwargs):
kwargs["noapi"] = app.config["pathod"].noapi
kwargs["nocraft"] = app.config["pathod"].nocraft
@ -37,30 +36,25 @@ def make_app(noapi):
resp.headers["Cache-control"] = "public, max-age=4320"
return resp
@app.route('/')
@app.route('/index.html')
def index():
return render("index.html", True, section="main")
@app.route('/download')
@app.route('/download.html')
def download():
return render("download.html", True, section="download", version=version.VERSION)
@app.route('/about')
@app.route('/about.html')
def about():
return render("about.html", True, section="about")
@app.route('/docs/pathod')
def docs_pathod():
return render("docs_pathod.html", True, section="docs", subsection="pathod")
@app.route('/docs/language')
def docs_language():
return render(
@ -69,29 +63,24 @@ def make_app(noapi):
subsection="lang"
)
@app.route('/docs/pathoc')
def docs_pathoc():
return render("docs_pathoc.html", True, section="docs", subsection="pathoc")
@app.route('/docs/libpathod')
def docs_libpathod():
return render("docs_libpathod.html", True, section="docs", subsection="libpathod")
@app.route('/docs/test')
def docs_test():
return render("docs_test.html", True, section="docs", subsection="test")
@app.route('/log')
def log():
if app.config["pathod"].noapi:
abort(404)
return render("log.html", False, section="log", log=app.config["pathod"].get_log())
@app.route('/log/<int:lid>')
def onelog(lid):
item = app.config["pathod"].log_by_id(int(lid))
@ -100,7 +89,6 @@ def make_app(noapi):
l = pprint.pformat(item)
return render("onelog.html", False, section="log", alog=l, lid=lid)
def _preview(is_request):
if is_request:
template = "request_preview.html"
@ -121,9 +109,9 @@ def make_app(noapi):
try:
if is_request:
r = language.parse_request(app.config["pathod"].request_settings, spec)
r = language.parse_requests(spec)[0]
else:
r = language.parse_response(app.config["pathod"].request_settings, spec)
r = language.parse_response(spec)
except language.ParseException, v:
args["syntaxerror"] = str(v)
args["marked"] = v.marked()
@ -144,14 +132,11 @@ def make_app(noapi):
args["output"] = utils.escape_unprintables(s.getvalue())
return render(template, False, **args)
@app.route('/response_preview')
def response_preview():
return _preview(False)
@app.route('/request_preview')
def request_preview():
return _preview(True)
return app

346
libpathod/main.py → libpathod/cmdline.py
View File

@ -1,10 +1,14 @@
#!/usr/bin/env python
import argparse, sys, logging, logging.handlers, os
from . import pathoc as _pathoc, pathod as _pathod, utils, version, language
from netlib import tcp, http_uastrings
import argparse
import os
import os.path
import sys
import re
from . import pathoc, pathod, version, utils, language
from netlib import http_uastrings
def pathoc():
def go_pathoc():
preparser = argparse.ArgumentParser(add_help=False)
preparser.add_argument(
"--show-uas", dest="showua", action="store_true", default=False,
@ -17,20 +21,41 @@ def pathoc():
print " ", i[1], i[0]
sys.exit(0)
parser = argparse.ArgumentParser(description='A perverse HTTP client.', parents=[preparser])
parser.add_argument('--version', action='version', version="pathoc " + version.VERSION)
parser = argparse.ArgumentParser(
description='A perverse HTTP client.', parents=[preparser]
)
parser.add_argument(
'--version',
action='version',
version="pathoc " + version.VERSION
)
parser.add_argument(
"-c", dest="connect_to", type=str, default=False,
metavar = "HOST:PORT",
help="Issue an HTTP CONNECT to connect to the specified host."
)
parser.add_argument(
"-n", dest='repeat', default=1, type=int, metavar="N",
help='Repeat requests N times'
"--memo-limit", dest='memolimit', default=5000, type=int, metavar="N",
help='Stop if we do not find a valid request after N attempts.'
)
parser.add_argument(
"-p", dest="port", type=int, default=None,
help="Port. Defaults to 80, or 443 if SSL is active"
"-m", dest='memo', action="store_true", default=False,
help="""
Remember specs, and never play the same one twice. Note that this
means requests have to be rendered in memory, which means that large
generated data can cause issues.
"""
)
parser.add_argument(
"-n", dest='repeat', default=1, type=int, metavar="N",
help='Repeat N times. If 0 repeat for ever.'
)
parser.add_argument(
"-r", dest="random", action="store_true", default=False,
help="""
Select a random request from those specified. If this is not specified,
requests are all played in sequence.
"""
)
parser.add_argument(
"-t", dest="timeout", type=int, default=None,
@ -38,13 +63,16 @@ def pathoc():
)
parser.add_argument(
'host', type=str,
help='Host to connect to'
metavar = "host[:port]",
help='Host and port to connect to'
)
parser.add_argument(
'request', type=str, nargs="+",
help='Request specification'
'requests', type=str, nargs="+",
help="""
Request specification, or path to a file containing request
specifcations
"""
)
group = parser.add_argument_group(
'SSL',
)
@ -67,7 +95,10 @@ def pathoc():
group.add_argument(
"--sslversion", dest="sslversion", type=int, default=4,
choices=[1, 2, 3, 4],
help="Use a specified protocol - TLSv1, SSLv2, SSLv3, SSLv23. Default to SSLv23."
help="""
Use a specified protocol - TLSv1, SSLv2, SSLv3, SSLv23. Default
to SSLv23.
"""
)
group = parser.add_argument_group(
@ -98,7 +129,7 @@ def pathoc():
help="Print full request"
)
group.add_argument(
"-r", dest="showresp", action="store_true", default=False,
"-p", dest="showresp", action="store_true", default=False,
help="Print full response"
)
group.add_argument(
@ -112,13 +143,21 @@ def pathoc():
args = parser.parse_args()
args.port = None
if ":" in args.host:
h, p = args.host.rsplit(":", 1)
try:
p = int(p)
except ValueError:
parser.error("Invalid port in host spec: %s" % args.host)
args.host = h
args.port = p
if args.port is None:
port = 443 if args.ssl else 80
else:
port = args.port
args.port = 443 if args.ssl else 80
try:
codes = [int(i) for i in args.ignorecodes.split(",") if i]
args.ignorecodes = [int(i) for i in args.ignorecodes.split(",") if i]
except ValueError:
parser.error("Invalid return code specification: %s"%args.ignorecodes)
@ -130,164 +169,59 @@ def pathoc():
parts[1] = int(parts[1])
except ValueError:
parser.error("Invalid CONNECT specification: %s"%args.connect_to)
connect_to = parts
args.connect_to = parts
else:
connect_to = None
args.connect_to = None
try:
for i in range(args.repeat):
p = _pathoc.Pathoc(
(args.host, port),
ssl=args.ssl,
sni=args.sni,
sslversion=args.sslversion,
clientcert=args.clientcert,
ciphers=args.ciphers
)
try:
p.connect(connect_to)
except (tcp.NetLibError, _pathoc.PathocError), v:
print >> sys.stderr, str(v)
sys.exit(1)
if args.timeout:
p.settimeout(args.timeout)
for spec in args.request:
ret = p.print_request(
spec,
showreq=args.showreq,
showresp=args.showresp,
explain=args.explain,
showssl=args.showssl,
hexdump=args.hexdump,
ignorecodes=codes,
ignoretimeout=args.ignoretimeout
)
sys.stdout.flush()
if ret and args.oneshot:
sys.exit(0)
except KeyboardInterrupt:
pass
def daemonize(stdin='/dev/null', stdout='/dev/null', stderr='/dev/null'):
try:
pid = os.fork()
if pid > 0:
sys.exit(0)
except OSError, e:
sys.stderr.write("fork #1 failed: (%d) %s\n" % (e.errno, e.strerror))
sys.exit(1)
os.chdir("/")
os.umask(0)
os.setsid()
try:
pid = os.fork()
if pid > 0:
sys.exit(0)
except OSError, e:
sys.stderr.write("fork #2 failed: (%d) %s\n" % (e.errno, e.strerror))
sys.exit(1)
si = open(stdin, 'rb')
so = open(stdout, 'a+b')
se = open(stderr, 'a+b', 0)
os.dup2(si.fileno(), sys.stdin.fileno())
os.dup2(so.fileno(), sys.stdout.fileno())
os.dup2(se.fileno(), sys.stderr.fileno())
def pathod_main(parser, args):
certs = []
for i in args.ssl_certs:
parts = i.split("=", 1)
if len(parts) == 1:
parts = ["*", parts[0]]
parts[1] = os.path.expanduser(parts[1])
if not os.path.exists(parts[1]):
parser.error("Certificate file does not exist: %s"%parts[1])
certs.append(parts)
ssloptions = _pathod.SSLOptions(
cn = args.cn,
confdir = args.confdir,
not_after_connect = args.ssl_not_after_connect,
ciphers = args.ciphers,
sslversion = utils.SSLVERSIONS[args.sslversion],
certs = certs
)
alst = []
for i in args.anchors:
parts = utils.parse_anchor_spec(i)
if not parts:
parser.error("Invalid anchor specification: %s"%i)
alst.append(parts)
root = logging.getLogger()
if root.handlers:
for handler in root.handlers:
root.removeHandler(handler)
log = logging.getLogger('pathod')
log.setLevel(logging.DEBUG)
fmt = logging.Formatter(
'%(asctime)s: %(message)s',
datefmt='%d-%m-%y %H:%M:%S',
)
if args.logfile:
fh = logging.handlers.WatchedFileHandler(args.logfile)
fh.setFormatter(fmt)
log.addHandler(fh)
if not args.daemonize:
sh = logging.StreamHandler()
sh.setFormatter(fmt)
log.addHandler(sh)
sizelimit = None
if args.sizelimit:
reqs = []
for r in args.requests:
if os.path.exists(r):
data = open(r).read()
r = data
try:
sizelimit = utils.parse_size(args.sizelimit)
except ValueError, v:
parser.error(v)
try:
pd = _pathod.Pathod(
(args.address, args.port),
craftanchor = args.craftanchor,
ssl = args.ssl,
ssloptions = ssloptions,
staticdir = args.staticdir,
anchors = alst,
sizelimit = sizelimit,
noweb = args.noweb,
nocraft = args.nocraft,
noapi = args.noapi,
nohang = args.nohang,
timeout = args.timeout,
logreq = args.logreq,
logresp = args.logresp,
hexdump = args.hexdump,
explain = args.explain,
)
except _pathod.PathodError, v:
parser.error(str(v))
except language.FileAccessDenied, v:
parser.error("%s You probably want to a -d argument."%str(v))
try:
print "%s listening on %s:%s"%(version.NAMEVERSION, pd.address.host, pd.address.port)
pd.serve_forever()
except KeyboardInterrupt:
pass
reqs.extend(language.parse_requests(r))
except language.ParseException, v:
print >> sys.stderr, "Error parsing request spec: %s"%v.msg
print >> sys.stderr, v.marked()
sys.exit(1)
args.requests = reqs
pathoc.main(args)
def pathod():
parser = argparse.ArgumentParser(description='A pathological HTTP/S daemon.')
parser.add_argument('--version', action='version', version="pathod " + version.VERSION)
parser.add_argument("-p", dest='port', default=9999, type=int, help='Port. Specify 0 to pick an arbitrary empty port.')
parser.add_argument("-l", dest='address', default="127.0.0.1", type=str, help='Listening address.')
def go_pathod():
parser = argparse.ArgumentParser(
description='A pathological HTTP/S daemon.'
)
parser.add_argument(
"-a", dest='anchors', default=[], type=str, action="append", metavar="ANCHOR",
help='Add an anchor. Specified as a string with the form pattern=pagespec'
'--version',
action='version',
version="pathod " + version.VERSION
)
parser.add_argument(
"-p",
dest='port',
default=9999,
type=int,
help='Port. Specify 0 to pick an arbitrary empty port.'
)
parser.add_argument(
"-l",
dest='address',
default="127.0.0.1",
type=str,
help='Listening address.'
)
parser.add_argument(
"-a",
dest='anchors',
default=[],
type=str,
action="append",
metavar="ANCHOR",
help="""
Add an anchor. Specified as a string with the form pattern=pagespec, or
pattern=filepath
"""
)
parser.add_argument(
"-c", dest='craftanchor', default="/p/", type=str,
@ -340,7 +274,7 @@ def pathod():
)
group.add_argument(
"--cn", dest="cn", type=str, default=None,
help="CN for generated SSL certs. Default: %s"%_pathod.DEFAULT_CERT_DOMAIN
help="CN for generated SSL certs. Default: %s"%pathod.DEFAULT_CERT_DOMAIN
)
group.add_argument(
"-C", dest='ssl_not_after_connect', default=False, action="store_true",
@ -349,10 +283,13 @@ def pathod():
group.add_argument(
"--cert", dest='ssl_certs', default=[], type=str,
metavar = "SPEC", action="append",
help='Add an SSL certificate. SPEC is of the form "[domain=]path". '\
'The domain may include a wildcard, and is equal to "*" if not specified. '\
'The file at path is a certificate in PEM format. If a private key is included in the PEM, '\
'it is used, else the default key in the conf dir is used. Can be passed multiple times.'
help = """
Add an SSL certificate. SPEC is of the form "[domain=]path". The domain
may include a wildcard, and is equal to "*" if not specified. The file
at path is a certificate in PEM format. If a private key is included in
the PEM, it is used, else the default key in the conf dir is used. Can
be passed multiple times.'
"""
)
group.add_argument(
"--ciphers", dest="ciphers", type=str, default=False,
@ -361,7 +298,8 @@ def pathod():
group.add_argument(
"--sslversion", dest="sslversion", type=int, default=4,
choices=[1, 2, 3, 4],
help="Use a specified protocol - TLSv1, SSLv2, SSLv3, SSLv23. Default to SSLv23."
help=""""Use a specified protocol - TLSv1, SSLv2, SSLv3, SSLv23. Default
to SSLv23."""
)
group = parser.add_argument_group(
@ -392,10 +330,52 @@ def pathod():
help="Log request/response in hexdump format"
)
args = parser.parse_args()
if args.daemonize:
daemonize()
pathod_main(parser, args)
certs = []
for i in args.ssl_certs:
parts = i.split("=", 1)
if len(parts) == 1:
parts = ["*", parts[0]]
parts[1] = os.path.expanduser(parts[1])
if not os.path.exists(parts[1]):
parser.error("Certificate file does not exist: %s"%parts[1])
certs.append(parts)
args.ssl_certs = certs
if __name__ == "__main__":
pathoc()
alst = []
for i in args.anchors:
parts = utils.parse_anchor_spec(i)
if not parts:
parser.error("Invalid anchor specification: %s"%i)
alst.append(parts)
args.anchors = alst
sizelimit = None
if args.sizelimit:
try:
sizelimit = utils.parse_size(args.sizelimit)
except ValueError, v:
parser.error(v)
args.sizelimit = sizelimit
anchors = []
for patt, spec in args.anchors:
if os.path.exists(spec):
data = open(spec).read()
spec = data
try:
req = language.parse_response(spec)
except language.ParseException, v:
print >> sys.stderr, "Error parsing anchor spec: %s"%v.msg
print >> sys.stderr, v.marked()
sys.exit(1)
try:
arex = re.compile(patt)
except re.error:
print >> sys.stderr, "Invalid regex in anchor: %s" % patt
sys.exit(1)
anchors.append((arex, req))
args.anchors = anchors
pathod.main(args)

755
libpathod/contrib/pyparsing.py
View File

File diff suppressed because it is too large Load Diff

225
libpathod/language.py
View File

@ -1,6 +1,11 @@
import operator, string, random, mmap, os, time, copy
import operator
import string
import random
import mmap
import os
import time
import copy
import abc
from email.utils import formatdate
import contrib.pyparsing as pp
from netlib import http_status, tcp, http_uastrings
@ -9,7 +14,20 @@ import utils
BLOCKSIZE = 1024
TRUNCATE = 1024
class FileAccessDenied(Exception): pass
def escape_backslash(s):
return s.replace("\\", "\\\\")
def quote(s):
quotechar = s[0]
s = s[1:-1]
s = s.replace(quotechar, "\\" + quotechar)
return quotechar + s + quotechar
class FileAccessDenied(Exception):
pass
class ParseException(Exception):
@ -20,7 +38,7 @@ class ParseException(Exception):
self.col = col
def marked(self):
return "%s\n%s"%(self.s, " "*(self.col-1) + "^")
return "%s\n%s"%(self.s, " "*(self.col - 1) + "^")
def __str__(self):
return "%s at char %s"%(self.msg, self.col)
@ -40,7 +58,9 @@ def send_chunk(fp, val, blocksize, start, end):
def write_values(fp, vals, actions, sofar=0, skip=0, blocksize=BLOCKSIZE):
"""
vals: A list of values, which may be strings or Value objects.
actions: A list of (offset, action, arg) tuples. Action may be "pause" or "disconnect".
actions: A list of (offset, action, arg) tuples. Action may be "pause"
or "disconnect".
Both vals and actions are in reverse order, with the first items last.
@ -53,7 +73,13 @@ def write_values(fp, vals, actions, sofar=0, skip=0, blocksize=BLOCKSIZE):
offset = 0
while actions and actions[-1][0] < (sofar + len(v)):
a = actions.pop()
offset += send_chunk(fp, v, blocksize, offset, a[0]-sofar-offset)
offset += send_chunk(
fp,
v,
blocksize,
offset,
a[0]-sofar-offset
)
if a[1] == "pause":
time.sleep(a[2])
elif a[1] == "disconnect":
@ -121,15 +147,25 @@ DATATYPES = dict(
)
v_integer = pp.Regex(r"\d+")\
v_integer = pp.Word(pp.nums)\
.setName("integer")\
.setParseAction(lambda toks: int(toks[0]))
v_literal = pp.MatchFirst(
[
pp.QuotedString("\"", escChar="\\", unquoteResults=True, multiline=True),
pp.QuotedString("'", escChar="\\", unquoteResults=True, multiline=True),
pp.QuotedString(
"\"",
escChar="\\",
unquoteResults=True,
multiline=True
),
pp.QuotedString(
"'",
escChar="\\",
unquoteResults=True,
multiline=True
),
]
)
@ -155,7 +191,7 @@ class LiteralGenerator:
return self.s.__getslice__(a, b)
def __repr__(self):
return '"%s"'%self.s
return "'%s'"%self.s
class RandomGenerator:
@ -202,6 +238,7 @@ class _Token(object):
A specification token. Tokens are immutable.
"""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def expr(klass): # pragma: no cover
"""
@ -242,10 +279,16 @@ class ValueLiteral(_ValueLiteral):
@classmethod
def expr(klass):
e = v_literal.copy()
return e.setParseAction(lambda x: klass(*x))
return e.setParseAction(klass.parseAction)
@classmethod
def parseAction(klass, x):
v = klass(*x)
return v
def spec(self):
return '"%s"'%self.val.encode("string_escape")
ret = "'%s'"%self.val.encode("string_escape")
return ret
class ValueNakedLiteral(_ValueLiteral):
@ -278,7 +321,10 @@ class ValueGenerate(_Token):
def expr(klass):
e = pp.Literal("@").suppress() + v_integer
u = reduce(operator.or_, [pp.Literal(i) for i in utils.SIZE_UNITS.keys()])
u = reduce(
operator.or_,
[pp.Literal(i) for i in utils.SIZE_UNITS.keys()]
).leaveWhitespace()
e = e + pp.Optional(u, default=None)
s = pp.Literal(",").suppress()
@ -318,13 +364,15 @@ class ValueFile(_Token):
s = os.path.expanduser(self.path)
s = os.path.normpath(os.path.abspath(os.path.join(sd, s)))
if not uf and not s.startswith(sd):
raise FileAccessDenied("File access outside of configured directory")
raise FileAccessDenied(
"File access outside of configured directory"
)
if not os.path.isfile(s):
raise FileAccessDenied("File not readable")
return FileGenerator(s)
def spec(self):
return '<"%s"'%self.path.encode("string_escape")
return "<'%s'"%self.path.encode("string_escape")
Value = pp.MatchFirst(
@ -347,12 +395,12 @@ NakedValue = pp.MatchFirst(
Offset = pp.MatchFirst(
[
v_integer,
pp.Literal("r"),
pp.Literal("a")
]
)
[
v_integer,
pp.Literal("r"),
pp.Literal("a")
]
)
class Raw(_Token):
@ -392,11 +440,11 @@ class _Header(_Component):
def values(self, settings):
return [
self.key.get_generator(settings),
": ",
self.value.get_generator(settings),
"\r\n",
]
self.key.get_generator(settings),
": ",
self.value.get_generator(settings),
"\r\n",
]
class Header(_Header):
@ -459,7 +507,10 @@ class ShortcutUserAgent(_Header):
@classmethod
def expr(klass):
e = pp.Literal("u").suppress()
u = reduce(operator.or_, [pp.Literal(i[1]) for i in http_uastrings.UASTRINGS])
u = reduce(
operator.or_,
[pp.Literal(i[1]) for i in http_uastrings.UASTRINGS]
)
e += u | Value
return e.setParseAction(lambda x: klass(*x))
@ -470,7 +521,6 @@ class ShortcutUserAgent(_Header):
return ShortcutUserAgent(self.value.freeze(settings))
class Body(_Component):
def __init__(self, value):
self.value = value
@ -483,8 +533,8 @@ class Body(_Component):
def values(self, settings):
return [
self.value.get_generator(settings),
]
self.value.get_generator(settings),
]
def spec(self):
return "b%s"%(self.value.spec())
@ -493,6 +543,38 @@ class Body(_Component):
return Body(self.value.freeze(settings))
class PathodSpec(_Token):
def __init__(self, value):
self.value = value
try:
self.parsed = Response(
Response.expr().parseString(
value.val,
parseAll=True
)
)
except pp.ParseException, v:
raise ParseException(v.msg, v.line, v.col)
@classmethod
def expr(klass):
e = pp.Literal("s").suppress()
e = e + ValueLiteral.expr()
return e.setParseAction(lambda x: klass(*x))
def values(self, settings):
return [
self.value.get_generator(settings),
]
def spec(self):
return "s%s"%(self.value.spec())
def freeze(self, settings):
f = self.parsed.freeze(settings).spec()
return PathodSpec(ValueLiteral(f.encode("string_escape")))
class Path(_Component):
def __init__(self, value):
if isinstance(value, basestring):
@ -506,8 +588,8 @@ class Path(_Component):
def values(self, settings):
return [
self.value.get_generator(settings),
]
self.value.get_generator(settings),
]
def spec(self):
return "%s"%(self.value.spec())
@ -527,6 +609,7 @@ class Method(_Component):
"trace",
"connect",
]
def __init__(self, value):
# If it's a string, we were passed one of the methods, so we upper-case
# it to be canonical. The user can specify a different case by using a
@ -645,11 +728,11 @@ class PauseAt(_Action):
e += Offset
e += pp.Literal(",").suppress()
e += pp.MatchFirst(
[
v_integer,
pp.Literal("f")
]
)
[
v_integer,
pp.Literal("f")
]
)
return e.setParseAction(lambda x: klass(*x))
def spec(self):
@ -700,10 +783,10 @@ class InjectAt(_Action):
def intermediate(self, settings):
return (
self.offset,
"inject",
self.value.get_generator(settings)
)
self.offset,
"inject",
self.value.get_generator(settings)
)
def freeze(self, settings):
return InjectAt(self.offset, self.value.freeze(settings))
@ -712,6 +795,7 @@ class InjectAt(_Action):
class _Message(object):
__metaclass__ = abc.ABCMeta
version = "HTTP/1.1"
def __init__(self, tokens):
self.tokens = tokens
@ -741,7 +825,8 @@ class _Message(object):
def length(self, settings):
"""
Calculate the length of the base message without any applied actions.
Calculate the length of the base message without any applied
actions.
"""
return sum(len(x) for x in self.values(settings))
@ -754,7 +839,8 @@ class _Message(object):
def maximum_length(self, settings):
"""
Calculate the maximum length of the base message with all applied actions.
Calculate the maximum length of the base message with all applied
actions.
"""
l = self.length(settings)
for i in self.actions:
@ -781,14 +867,6 @@ class _Message(object):
ValueLiteral(request_host)
)
)
else:
if not utils.get_header("Date", self.headers):
tokens.append(
Header(
ValueLiteral("Date"),
ValueLiteral(formatdate(timeval=None, localtime=False, usegmt=True))
)
)
intermediate = self.__class__(tokens)
return self.__class__([i.resolve(intermediate, settings) for i in tokens])
@ -807,7 +885,8 @@ class _Message(object):
ret = {}
for i in self.logattrs:
v = getattr(self, i)
# Careful not to log any VALUE specs without sanitizing them first. We truncate at 1k.
# Careful not to log any VALUE specs without sanitizing them first.
# We truncate at 1k.
if hasattr(v, "values"):
v = [x[:TRUNCATE] for x in v.values(settings)]
v = "".join(v).encode("string_escape")
@ -838,6 +917,7 @@ class _Message(object):
Sep = pp.Optional(pp.Literal(":")).suppress()
class Response(_Message):
comps = (
Body,
@ -851,6 +931,7 @@ class Response(_Message):
Reason
)
logattrs = ["code", "reason", "version", "body"]
@property
def code(self):
return self._get_token(Code)
@ -866,7 +947,14 @@ class Response(_Message):
if self.reason:
l.extend(self.reason.values(settings))
else:
l.append(LiteralGenerator(http_status.RESPONSES.get(int(self.code.code), "Unknown code")))
l.append(
LiteralGenerator(
http_status.RESPONSES.get(
int(self.code.code),
"Unknown code"
)
)
)
return l
@classmethod
@ -894,9 +982,11 @@ class Request(_Message):
InjectAt,
ShortcutContentType,
ShortcutUserAgent,
Raw
Raw,
PathodSpec,
)
logattrs = ["method", "path", "body"]
@property
def method(self):
return self._get_token(Method)
@ -905,10 +995,16 @@ class Request(_Message):
def path(self):
return self._get_token(Path)
@property
def pathodspec(self):
return self._get_token(PathodSpec)
def preamble(self, settings):
v = self.method.values(settings)
v.append(" ")
v.extend(self.path.values(settings))
if self.pathodspec:
v.append(self.pathodspec.parsed.spec())
v.append(" ")
v.append(self.version)
return v
@ -944,7 +1040,7 @@ def make_error_response(reason, body=None):
]
return PathodErrorResponse(tokens)
FILESTART = "+"
def read_file(settings, s):
uf = settings.get("unconstrained_file_access")
sd = settings.get("staticdir")
@ -961,33 +1057,34 @@ def read_file(settings, s):
return file(s, "rb").read()
def parse_response(settings, s):
def parse_response(s):
"""
May raise ParseException or FileAccessDenied
May raise ParseException
"""
try:
s = s.decode("ascii")
except UnicodeError:
raise ParseException("Spec must be valid ASCII.", 0, 0)
if s.startswith(FILESTART):
s = read_file(settings, s)
try:
return Response(Response.expr().parseString(s, parseAll=True))
except pp.ParseException, v:
raise ParseException(v.msg, v.line, v.col)
def parse_request(settings, s):
def parse_requests(s):
"""
May raise ParseException or FileAccessDenied
May raise ParseException
"""
try:
s = s.decode("ascii")
except UnicodeError:
raise ParseException("Spec must be valid ASCII.", 0, 0)
if s.startswith(FILESTART):
s = read_file(settings, s)
try:
return Request(Request.expr().parseString(s, parseAll=True))
parts = pp.OneOrMore(
pp.Group(
Request.expr()
)
).parseString(s, parseAll=True)
return [Request(i) for i in parts]
except pp.ParseException, v:
raise ParseException(v.msg, v.line, v.col)

170
libpathod/pathoc.py
View File

@ -1,10 +1,17 @@
import sys, os
import sys
import os
import hashlib
import random
from netlib import tcp, http, certutils
import netlib.utils
import language, utils
import language
import utils
import OpenSSL.crypto
class PathocError(Exception): pass
class PathocError(Exception):
pass
class SSLInfo:
@ -13,8 +20,17 @@ class SSLInfo:
class Response:
def __init__(self, httpversion, status_code, msg, headers, content, sslinfo):
self.httpversion, self.status_code, self.msg = httpversion, status_code, msg
def __init__(
self,
httpversion,
status_code,
msg,
headers,
content,
sslinfo
):
self.httpversion, self.status_code = httpversion, status_code
self.msg = msg
self.headers, self.content = headers, content
self.sslinfo = sslinfo
@ -23,7 +39,14 @@ class Response:
class Pathoc(tcp.TCPClient):
def __init__(self, address, ssl=None, sni=None, sslversion=4, clientcert=None, ciphers=None):
def __init__(
self,
address,
ssl=None,
sni=None,
sslversion=4,
clientcert=None,
ciphers=None):
tcp.TCPClient.__init__(self, address)
self.settings = dict(
staticdir = os.getcwd(),
@ -36,9 +59,9 @@ class Pathoc(tcp.TCPClient):
def http_connect(self, connect_to):
self.wfile.write(
'CONNECT %s:%s HTTP/1.1\r\n'%tuple(connect_to) +
'\r\n'
)
'CONNECT %s:%s HTTP/1.1\r\n'%tuple(connect_to) +
'\r\n'
)
self.wfile.flush()
l = self.rfile.readline()
if not l:
@ -60,17 +83,17 @@ class Pathoc(tcp.TCPClient):
if self.ssl:
try:
self.convert_to_ssl(
sni=self.sni,
cert=self.clientcert,
method=self.sslversion,
cipher_list = self.ciphers
)
sni=self.sni,
cert=self.clientcert,
method=self.sslversion,
cipher_list = self.ciphers
)
except tcp.NetLibError, v:
raise PathocError(str(v))
self.sslinfo = SSLInfo(
self.connection.get_peer_cert_chain(),
self.get_current_cipher()
)
self.connection.get_peer_cert_chain(),
self.get_current_cipher()
)
def request(self, spec):
"""
@ -79,7 +102,7 @@ class Pathoc(tcp.TCPClient):
May raise language.ParseException, netlib.http.HttpError or
language.FileAccessDenied.
"""
r = language.parse_request(self.settings, spec)
r = language.parse_requests(spec)[0]
language.serve(r, self.wfile, self.settings, self.address.host)
self.wfile.flush()
ret = list(http.read_response(self.rfile, r.method.string(), None))
@ -87,7 +110,9 @@ class Pathoc(tcp.TCPClient):
return Response(*ret)
def _show_summary(self, fp, httpversion, code, msg, headers, content):
print >> fp, "<< %s %s: %s bytes"%(code, utils.xrepr(msg), len(content))
print >> fp, "<< %s %s: %s bytes"%(
code, utils.xrepr(msg), len(content)
)
def _show(self, fp, header, data, hexdump):
if hexdump:
@ -98,7 +123,18 @@ class Pathoc(tcp.TCPClient):
print >> fp, "%s (unprintables escaped):"%header
print >> fp, netlib.utils.cleanBin(data)
def print_request(self, spec, showreq, showresp, explain, showssl, hexdump, ignorecodes, ignoretimeout, fp=sys.stdout):
def print_request(
self,
r,
showreq,
showresp,
explain,
showssl,
hexdump,
ignorecodes,
ignoretimeout,
fp=sys.stdout
):
"""
Performs a series of requests, and prints results to the specified
file descriptor.
@ -113,26 +149,18 @@ class Pathoc(tcp.TCPClient):
Returns True if we have a non-ignored response.
"""
try:
r = language.parse_request(self.settings, spec)
except language.ParseException, v:
print >> fp, "Error parsing request spec: %s"%v.msg
print >> fp, v.marked()
return
except language.FileAccessDenied, v:
print >> fp, "File access error: %s"%v
return
if explain:
r = r.freeze(self.settings, self.address.host)
resp, req = None, None
if showreq:
self.wfile.start_log()
if showresp:
self.rfile.start_log()
try:
req = language.serve(r, self.wfile, self.settings, self.address.host)
req = language.serve(
r,
self.wfile,
self.settings,
self.address.host
)
self.wfile.flush()
resp = http.read_response(self.rfile, r.method.string(), None)
except http.HttpError, v:
@ -160,7 +188,7 @@ class Pathoc(tcp.TCPClient):
if resp:
self._show_summary(fp, *resp)
if self.sslinfo:
if showssl and self.sslinfo:
print >> fp, "Cipher: %s, %s bit, %s"%self.sslinfo.cipher
print >> fp, "SSL certificate chain:\n"
for i in self.sslinfo.certchain:
@ -173,13 +201,15 @@ class Pathoc(tcp.TCPClient):
print >> fp, "%s=%s"%cn,
print >> fp
print >> fp, "\tVersion: %s"%i.get_version()
print >> fp, "\tValidity: %s - %s"%(i.get_notBefore(),i.get_notAfter())
print >> fp, "\tValidity: %s - %s"%(
i.get_notBefore(), i.get_notAfter()
)
print >> fp, "\tSerial: %s"%i.get_serial_number()
print >> fp, "\tAlgorithm: %s"%i.get_signature_algorithm()
pk = i.get_pubkey()
types = {
OpenSSL.crypto.TYPE_RSA: "RSA",
OpenSSL.crypto.TYPE_DSA: "DSA"
OpenSSL.crypto.TYPE_RSA: "RSA",
OpenSSL.crypto.TYPE_DSA: "DSA"
}
t = types.get(pk.type(), "Uknown")
print >> fp, "\tPubkey: %s bit %s"%(pk.bits(), t)
@ -190,4 +220,68 @@ class Pathoc(tcp.TCPClient):
return True
def main(args):
memo = set([])
trycount = 0
try:
cnt = 0
while 1:
if trycount > args.memolimit:
print >> sys.stderr, "Memo limit exceeded..."
return
cnt += 1
if args.random:
playlist = [random.choice(args.requests)]
else:
playlist = args.requests
p = Pathoc(
(args.host, args.port),
ssl=args.ssl,
sni=args.sni,
sslversion=args.sslversion,
clientcert=args.clientcert,
ciphers=args.ciphers
)
if args.explain or args.memo:
playlist = [
i.freeze(p.settings, p.address.host) for i in playlist
]
if args.memo:
newlist = []
for spec in playlist:
h = hashlib.sha256(spec.spec()).digest()
if h not in memo:
memo.add(h)
newlist.append(spec)
playlist = newlist
if not playlist:
trycount += 1
continue
trycount = 0
try:
p.connect(args.connect_to)
except (tcp.NetLibError, PathocError), v:
print >> sys.stderr, str(v)
sys.exit(1)
if args.timeout:
p.settimeout(args.timeout)
for spec in playlist:
ret = p.print_request(
spec,
showreq=args.showreq,
showresp=args.showresp,
explain=args.explain,
showssl=args.showssl,
hexdump=args.hexdump,
ignorecodes=args.ignorecodes,
ignoretimeout=args.ignoretimeout
)
sys.stdout.flush()
if ret and args.oneshot:
sys.exit(0)
if cnt == args.repeat:
break
except KeyboardInterrupt:
pass

146
libpathod/pathod.py
View File

@ -1,7 +1,12 @@
import urllib, threading, re, logging, os
import urllib
import threading
import logging
import os
import sys
from netlib import tcp, http, wsgi, certutils
import netlib.utils
import version, app, language, utils
from . import version, app, language, utils
DEFAULT_CERT_DOMAIN = "pathod.net"
@ -12,7 +17,8 @@ CA_CERT_NAME = "mitmproxy-ca.pem"
logger = logging.getLogger('pathod')
class PathodError(Exception): pass
class PathodError(Exception):
pass
class SSLOptions:
@ -64,7 +70,12 @@ class PathodHandler(tcp.BaseHandler):
if self.server.explain and not isinstance(crafted, language.PathodErrorResponse):
crafted = crafted.freeze(self.server.request_settings, None)
self.info(">> Spec: %s" % crafted.spec())
response_log = language.serve(crafted, self.wfile, self.server.request_settings, None)
response_log = language.serve(
crafted,
self.wfile,
self.server.request_settings,
None
)
if response_log["disconnect"]:
return False, response_log
return True, response_log
@ -162,15 +173,14 @@ class PathodHandler(tcp.BaseHandler):
for i in self.server.anchors:
if i[0].match(path):
self.info("crafting anchor: %s" % path)
aresp = language.parse_response(self.server.request_settings, i[1])
again, retlog["response"] = self.serve_crafted(aresp)
again, retlog["response"] = self.serve_crafted(i[1])
return again, retlog
if not self.server.nocraft and path.startswith(self.server.craftanchor):
spec = urllib.unquote(path)[len(self.server.craftanchor):]
self.info("crafting spec: %s" % spec)
try:
crafted = language.parse_response(self.server.request_settings, spec)
crafted = language.parse_response(spec)
except language.ParseException, v:
self.info("Parse error: %s" % v.msg)
crafted = language.make_error_response(
@ -182,7 +192,10 @@ class PathodHandler(tcp.BaseHandler):
elif self.server.noweb:
crafted = language.make_error_response("Access Denied")
language.serve(crafted, self.wfile, self.server.request_settings)
return False, dict(type="error", msg="Access denied: web interface disabled")
return False, dict(
type="error",
msg="Access denied: web interface disabled"
)
else:
self.info("app: %s %s" % (method, path))
req = wsgi.Request("http", method, path, headers, content)
@ -252,19 +265,34 @@ class Pathod(tcp.TCPServer):
LOGBUF = 500
def __init__(
self, addr, confdir=CONFDIR, ssl=False, ssloptions=None,
craftanchor="/p/", staticdir=None, anchors=None,
sizelimit=None, noweb=False, nocraft=False, noapi=False,
nohang=False, timeout=None, logreq=False, logresp=False,
explain=False, hexdump=False
self,
addr,
confdir=CONFDIR,
ssl=False,
ssloptions=None,
craftanchor="/p/",
staticdir=None,
anchors=(),
sizelimit=None,
noweb=False,
nocraft=False,
noapi=False,
nohang=False,
timeout=None,
logreq=False,
logresp=False,
explain=False,
hexdump=False
):
"""
addr: (address, port) tuple. If port is 0, a free port will be
automatically chosen.
ssloptions: an SSLOptions object.
craftanchor: string specifying the path under which to anchor response generation.
craftanchor: string specifying the path under which to anchor
response generation.
staticdir: path to a directory of static resources, or None.
anchors: A list of (regex, spec) tuples, or None.
anchors: List of (regex object, language.Request object) tuples, or
None.
sizelimit: Limit size of served data.
nocraft: Disable response crafting.
noapi: Disable the API.
@ -276,26 +304,17 @@ class Pathod(tcp.TCPServer):
self.staticdir = staticdir
self.craftanchor = craftanchor
self.sizelimit = sizelimit
self.noweb, self.nocraft, self.noapi, self.nohang = noweb, nocraft, noapi, nohang
self.timeout, self.logreq, self.logresp, self.hexdump = timeout, logreq, logresp, hexdump
self.noweb, self.nocraft = noweb, nocraft
self.noapi, self.nohang = noapi, nohang
self.timeout, self.logreq = timeout, logreq
self.logresp, self.hexdump = logresp, hexdump
self.explain = explain
self.app = app.make_app(noapi)
self.app.config["pathod"] = self
self.log = []
self.logid = 0
self.anchors = []
if anchors:
for i in anchors:
try:
arex = re.compile(i[0])
except re.error:
raise PathodError("Invalid regex in anchor: %s" % i[0])
try:
language.parse_response(self.request_settings, i[1])
except language.ParseException, v:
raise PathodError("Invalid page spec in anchor: '%s', %s" % (i[1], str(v)))
self.anchors.append((arex, i[1]))
self.anchors = anchors
def check_policy(self, req, settings):
"""
@ -363,3 +382,72 @@ class Pathod(tcp.TCPServer):
def get_log(self):
return self.log
def main(args):
ssloptions = SSLOptions(
cn = args.cn,
confdir = args.confdir,
not_after_connect = args.ssl_not_after_connect,
ciphers = args.ciphers,
sslversion = utils.SSLVERSIONS[args.sslversion],
certs = args.ssl_certs
)
root = logging.getLogger()
if root.handlers:
for handler in root.handlers:
root.removeHandler(handler)
log = logging.getLogger('pathod')
log.setLevel(logging.DEBUG)
fmt = logging.Formatter(
'%(asctime)s: %(message)s',
datefmt='%d-%m-%y %H:%M:%S',
)
if args.logfile:
fh = logging.handlers.WatchedFileHandler(args.logfile)
fh.setFormatter(fmt)
log.addHandler(fh)
if not args.daemonize:
sh = logging.StreamHandler()
sh.setFormatter(fmt)
log.addHandler(sh)
try:
pd = Pathod(
(args.address, args.port),
craftanchor = args.craftanchor,
ssl = args.ssl,
ssloptions = ssloptions,
staticdir = args.staticdir,
anchors = args.anchors,
sizelimit = args.sizelimit,
noweb = args.noweb,
nocraft = args.nocraft,
noapi = args.noapi,
nohang = args.nohang,
timeout = args.timeout,
logreq = args.logreq,
logresp = args.logresp,
hexdump = args.hexdump,
explain = args.explain,
)
except PathodError, v:
print >> sys.stderr, "Error: %s"%v
sys.exit(1)
except language.FileAccessDenied, v:
print >> sys.stderr, "Error: %s"%v
if args.daemonize:
utils.daemonize()
try:
print "%s listening on %s:%s"%(
version.NAMEVERSION,
pd.address.host,
pd.address.port
)
pd.serve_forever()
except KeyboardInterrupt:
pass

8
libpathod/templates/docs_lang.html
View File

@ -183,14 +183,6 @@
</div>
<section id="specifying_requests">
<div class="page-header">
<h1>Executing specs from file</h1>
</div>
<pre class="example">+./path/to/spec</pre>
</section>
<section id="specifying_requests">
<div class="page-header">

15
libpathod/test.py
View File

@ -1,4 +1,5 @@
import threading, Queue
import threading
import Queue
import requests
import requests.packages.urllib3
import pathod
@ -6,7 +7,6 @@ import pathod
requests.packages.urllib3.disable_warnings()
class Daemon:
IFACE = "127.0.0.1"
def __init__(self, ssl=None, **daemonargs):
@ -14,7 +14,11 @@ class Daemon:
self.thread = _PaThread(self.IFACE, self.q, ssl, daemonargs)
self.thread.start()
self.port = self.q.get(True, 5)
self.urlbase = "%s://%s:%s"%("https" if ssl else "http", self.IFACE, self.port)
self.urlbase = "%s://%s:%s"%(
"https" if ssl else "http",
self.IFACE,
self.port
)
def __enter__(self):
return self
@ -80,6 +84,9 @@ class _PaThread(threading.Thread):
ssl = self.ssl,
**self.daemonargs
)
self.name = "PathodThread (%s:%s)" % (self.server.address.host, self.server.address.port)
self.name = "PathodThread (%s:%s)" % (
self.server.address.host,
self.server.address.port
)
self.q.put(self.server.address.port)
self.server.serve_forever()

28
libpathod/utils.py
View File

@ -1,4 +1,5 @@
import os
import sys
from netlib import tcp
SSLVERSIONS = {
@ -110,3 +111,30 @@ class Data:
data = Data(__name__)
def daemonize(stdin='/dev/null', stdout='/dev/null', stderr='/dev/null'):
try:
pid = os.fork()
if pid > 0:
sys.exit(0)
except OSError, e:
sys.stderr.write("fork #1 failed: (%d) %s\n" % (e.errno, e.strerror))
sys.exit(1)
os.chdir("/")
os.umask(0)
os.setsid()
try:
pid = os.fork()
if pid > 0:
sys.exit(0)
except OSError, e:
sys.stderr.write("fork #2 failed: (%d) %s\n" % (e.errno, e.strerror))
sys.exit(1)
si = open(stdin, 'rb')
so = open(stdout, 'a+b')
se = open(stderr, 'a+b', 0)
os.dup2(si.fileno(), sys.stdin.fileno())
os.dup2(so.fileno(), sys.stdout.fileno())
os.dup2(se.fileno(), sys.stderr.fileno())

5
pathoc Executable file
View File

@ -0,0 +1,5 @@
#!/usr/bin/env python
from libpathod import cmdline
if __name__ == "__main__":
cmdline.go_pathoc()

5
pathod Executable file
View File

@ -0,0 +1,5 @@
#!/usr/bin/env python
from libpathod import cmdline
if __name__ == "__main__":
cmdline.go_pathod()

12
setup.py
View File

@ -35,16 +35,12 @@ setup(
packages=find_packages(),
include_package_data=True,
entry_points={
'console_scripts': [
"pathod = libpathod.main:pathod",
"pathoc = libpathod.main:pathoc"
]
},
scripts = ["pathod", "pathoc"],
install_requires=[
"netlib>=%s, <%s" % (version.MINORVERSION, version.NEXT_MINORVERSION),
# It's INSANE that we have to do this, but...
# FIXME: Requirement to be removed at next release
"pip>=1.5.6",
"requests>=2.4.1",
"Flask>=0.10.1"
],

3
test/test_app.py
View File

@ -1,7 +1,9 @@
import tutils
class TestApp(tutils.DaemonTests):
SSL = False
def test_index(self):
r = self.getpath("/")
assert r.status_code == 200
@ -78,4 +80,3 @@ class TestApp(tutils.DaemonTests):
r = self.getpath("/request_preview", params=dict(spec=""))
assert r.status_code == 200
assert 'empty spec' in r.content

255
test/test_language.py
View File

@ -1,10 +1,19 @@
import os, cStringIO
import os
import cStringIO
from libpathod import language, utils
import tutils
language.TESTING = True
def test_quote():
assert language.quote("'\\\\'")
def parse_request(s):
return language.parse_requests(s)[0]
class TestValueNakedLiteral:
def test_expr(self):
v = language.ValueNakedLiteral("foo")
@ -24,21 +33,35 @@ class TestValueLiteral:
assert v.expr()
assert v.val == "foo"
v = language.ValueLiteral(r"foo\n")
v = language.ValueLiteral("foo\n")
assert v.expr()
assert v.val == "foo\n"
assert repr(v)
def test_spec(self):
v = language.ValueLiteral("foo")
assert v.spec() == r'"foo"'
assert v.spec() == r"'foo'"
v = language.ValueLiteral("f\x00oo")
assert v.spec() == repr(v) == r'"f\x00oo"'
assert v.spec() == repr(v) == r"'f\x00oo'"
def test_freeze(self):
v = language.ValueLiteral("foo")
assert v.freeze({}).val == v.val
v = language.ValueLiteral("\"")
assert v.spec() == repr(v) == '\'"\''
def roundtrip(self, spec):
e = language.ValueLiteral.expr()
v = language.ValueLiteral(spec)
v2 = e.parseString(v.spec())
assert v.val == v2[0].val
assert v.spec() == v2[0].spec()
def test_roundtrip(self):
self.roundtrip("'")
self.roundtrip('\'')
self.roundtrip("a")
self.roundtrip("\"")
self.roundtrip(r"\\")
self.roundtrip("200:b'foo':i23,'\\''")
class TestValueGenerate:
@ -181,7 +204,7 @@ class TestMisc:
assert e.parseString("'get'")[0].value.val == "get"
assert e.parseString("get")[0].spec() == "get"
assert e.parseString("'foo'")[0].spec() == '"foo"'
assert e.parseString("'foo'")[0].spec() == "'foo'"
s = e.parseString("get")[0].spec()
assert s == e.parseString(s)[0].spec()
@ -218,6 +241,31 @@ class TestMisc:
s = v.spec()
assert s == e.parseString(s)[0].spec()
def test_pathodspec(self):
e = language.PathodSpec.expr()
v = e.parseString("s'200'")[0]
assert v.value.val == "200"
tutils.raises(
language.ParseException,
e.parseString,
"s'foo'"
)
v = e.parseString('s"200:b@1"')[0]
assert "@1" in v.spec()
f = v.freeze({})
assert "@1" not in f.spec()
def test_pathodspec_freeze(self):
e = language.PathodSpec(
language.ValueLiteral(
"200:b'foo':i10,'\\''".encode(
"string_escape"
)
)
)
assert e.freeze({})
def test_code(self):
e = language.Code.expr()
v = e.parseString("200")[0]
@ -298,11 +346,11 @@ class TestHeaders:
assert v2.value.val == v3.value.val
def test_shortcuts(self):
assert language.parse_response({}, "400:c'foo'").headers[0].key.val == "Content-Type"
assert language.parse_response({}, "400:l'foo'").headers[0].key.val == "Location"
assert language.parse_response("400:c'foo'").headers[0].key.val == "Content-Type"
assert language.parse_response("400:l'foo'").headers[0].key.val == "Location"
assert 'Android' in language.parse_request({}, "get:/:ua").headers[0].value.val
assert language.parse_request({}, "get:/:ua").headers[0].key.val == "User-Agent"
assert 'Android' in parse_request("get:/:ua").headers[0].value.val
assert parse_request("get:/:ua").headers[0].key.val == "User-Agent"
class TestShortcutUserAgent:
@ -336,7 +384,7 @@ class Test_Action:
assert l[0].offset == 0
def test_resolve(self):
r = language.parse_request({}, 'GET:"/foo"')
r = parse_request('GET:"/foo"')
e = language.DisconnectAt("r")
ret = e.resolve(r, {})
assert isinstance(ret.offset, int)
@ -352,9 +400,9 @@ class Test_Action:
class TestDisconnects:
def test_parse_response(self):
a = language.parse_response({}, "400:d0").actions[0]
a = language.parse_response("400:d0").actions[0]
assert a.spec() == "d0"
a = language.parse_response({}, "400:dr").actions[0]
a = language.parse_response("400:dr").actions[0]
assert a.spec() == "dr"
def test_at(self):
@ -377,12 +425,12 @@ class TestDisconnects:
class TestInject:
def test_parse_response(self):
a = language.parse_response({}, "400:ir,@100").actions[0]
a = language.parse_response("400:ir,@100").actions[0]
assert a.offset == "r"
assert a.value.datatype == "bytes"
assert a.value.usize == 100
a = language.parse_response({}, "400:ia,@100").actions[0]
a = language.parse_response("400:ia,@100").actions[0]
assert a.offset == "a"
def test_at(self):
@ -397,7 +445,7 @@ class TestInject:
def test_serve(self):
s = cStringIO.StringIO()
r = language.parse_response({}, "400:i0,'foo'")
r = language.parse_response("400:i0,'foo'")
assert language.serve(r, s, {})
def test_spec(self):
@ -430,7 +478,7 @@ class TestPauses:
assert v.offset == "a"
def test_request(self):
r = language.parse_response({}, '400:p10,10')
r = language.parse_response('400:p10,10')
assert r.actions[0].spec() == "p10,10"
def test_spec(self):
@ -444,30 +492,59 @@ class TestPauses:
class TestRequest:
def test_file(self):
p = tutils.test_data.path("data")
d = dict(staticdir=p)
r = language.parse_request(d, "+request")
assert r.path.values({})[0][:] == "/foo"
def test_nonascii(self):
tutils.raises("ascii", language.parse_request, {}, "get:\xf0")
tutils.raises("ascii", parse_request, "get:\xf0")
def test_err(self):
tutils.raises(language.ParseException, language.parse_request, {}, 'GET')
tutils.raises(language.ParseException, parse_request, 'GET')
def test_simple(self):
r = language.parse_request({}, 'GET:"/foo"')
r = parse_request('GET:"/foo"')
assert r.method.string() == "GET"
assert r.path.string() == "/foo"
r = language.parse_request({}, 'GET:/foo')
r = parse_request('GET:/foo')
assert r.path.string() == "/foo"
r = language.parse_request({}, 'GET:@1k')
r = parse_request('GET:@1k')
assert len(r.path.string()) == 1024
def test_multiple(self):
r = language.parse_requests("GET:/ PUT:/")
assert r[0].method.string() == "GET"
assert r[1].method.string() == "PUT"
assert len(r) == 2
l = """
GET
"/foo"
ir,@1
PUT
"/foo
bar"
ir,@1
"""
r = language.parse_requests(l)
assert len(r) == 2
assert r[0].method.string() == "GET"
assert r[1].method.string() == "PUT"
l = """
get:"http://localhost:9999/p/200":ir,@1
get:"http://localhost:9999/p/200":ir,@2
"""
r = language.parse_requests(l)
assert len(r) == 2
assert r[0].method.string() == "GET"
assert r[1].method.string() == "GET"
def test_render(self):
s = cStringIO.StringIO()
r = language.parse_request({}, "GET:'/foo'")
r = parse_request("GET:'/foo'")
assert language.serve(r, s, {}, "foo.com")
def test_multiline(self):
@ -476,12 +553,11 @@ class TestRequest:
"/foo"
ir,@1
"""
r = language.parse_request({}, l)
r = parse_request(l)
assert r.method.string() == "GET"
assert r.path.string() == "/foo"
assert r.actions
l = """
GET
@ -493,24 +569,24 @@ class TestRequest:
ir,@1
"""
r = language.parse_request({}, l)
r = parse_request(l)
assert r.method.string() == "GET"
assert r.path.string().endswith("bar")
assert r.actions
def test_spec(self):
def rt(s):
s = language.parse_request({}, s).spec()
assert language.parse_request({}, s).spec() == s
s = parse_request(s).spec()
assert parse_request(s).spec() == s
rt("get:/foo")
rt("get:/foo:da")
def test_freeze(self):
r = language.parse_request({}, "GET:/:b@100").freeze({})
r = parse_request("GET:/:b@100").freeze({})
assert len(r.spec()) > 100
def test_path_generator(self):
r = language.parse_request({}, "GET:@100").freeze({})
r = parse_request("GET:@100").freeze({})
assert len(r.spec()) > 100
@ -580,72 +656,69 @@ class TestWriteValues:
def test_write_values_after(self):
s = cStringIO.StringIO()
r = language.parse_response({}, "400:da")
r = language.parse_response("400:da")
language.serve(r, s, {})
s = cStringIO.StringIO()
r = language.parse_response({}, "400:pa,0")
r = language.parse_response("400:pa,0")
language.serve(r, s, {})
s = cStringIO.StringIO()
r = language.parse_response({}, "400:ia,'xx'")
r = language.parse_response("400:ia,'xx'")
language.serve(r, s, {})
assert s.getvalue().endswith('xx')
class TestResponse:
def dummy_response(self):
return language.parse_response({}, "400'msg'")
def test_file(self):
p = tutils.test_data.path("data")
d = dict(staticdir=p)
r = language.parse_response(d, "+response")
assert r.code.string() == "202"
return language.parse_response("400'msg'")
def test_response(self):
r = language.parse_response({}, "400:m'msg'")
r = language.parse_response("400:m'msg'")
assert r.code.string() == "400"
assert r.reason.string() == "msg"
r = language.parse_response({}, "400:m'msg':b@100b")
r = language.parse_response("400:m'msg':b@100b")
assert r.reason.string() == "msg"
assert r.body.values({})
assert str(r)
r = language.parse_response({}, "200")
r = language.parse_response("200")
assert r.code.string() == "200"
assert not r.reason
assert "OK" in [i[:] for i in r.preamble({})]
def test_render(self):
s = cStringIO.StringIO()
r = language.parse_response({}, "400:m'msg'")
r = language.parse_response("400:m'msg'")
assert language.serve(r, s, {})
r = language.parse_response("400:p0,100:dr")
assert "p0" in r.spec()
s = r.preview_safe()
assert "p0" not in s.spec()
def test_raw(self):
s = cStringIO.StringIO()
r = language.parse_response({}, "400:b'foo'")
r = language.parse_response("400:b'foo'")
language.serve(r, s, {})
v = s.getvalue()
assert "Content-Length" in v
assert "Date" in v
s = cStringIO.StringIO()
r = language.parse_response({}, "400:b'foo':r")
r = language.parse_response("400:b'foo':r")
language.serve(r, s, {})
v = s.getvalue()
assert not "Content-Length" in v
assert not "Date" in v
def test_length(self):
def testlen(x):
s = cStringIO.StringIO()
language.serve(x, s, {})
assert x.length({}) == len(s.getvalue())
testlen(language.parse_response({}, "400:m'msg':r"))
testlen(language.parse_response({}, "400:m'msg':h'foo'='bar':r"))
testlen(language.parse_response({}, "400:m'msg':h'foo'='bar':b@100b:r"))
testlen(language.parse_response("400:m'msg':r"))
testlen(language.parse_response("400:m'msg':h'foo'='bar':r"))
testlen(language.parse_response("400:m'msg':h'foo'='bar':b@100b:r"))
def test_maximum_length(self):
def testlen(x):
@ -654,74 +727,84 @@ class TestResponse:
language.serve(x, s, {})
assert m >= len(s.getvalue())
r = language.parse_response({}, "400:m'msg':b@100:d0")
r = language.parse_response("400:m'msg':b@100:d0")
testlen(r)
r = language.parse_response({}, "400:m'msg':b@100:d0:i0,'foo'")
r = language.parse_response("400:m'msg':b@100:d0:i0,'foo'")
testlen(r)
r = language.parse_response({}, "400:m'msg':b@100:d0:i0,'foo'")
r = language.parse_response("400:m'msg':b@100:d0:i0,'foo'")
testlen(r)
def test_render(self):
r = language.parse_response({}, "400:p0,100:dr")
assert "p0" in r.spec()
s = r.preview_safe()
assert not "p0" in s.spec()
def test_parse_err(self):
tutils.raises(language.ParseException, language.parse_response, {}, "400:msg,b:")
tutils.raises(
language.ParseException, language.parse_response, "400:msg,b:"
)
try:
language.parse_response({}, "400'msg':b:")
language.parse_response("400'msg':b:")
except language.ParseException, v:
assert v.marked()
assert str(v)
def test_nonascii(self):
tutils.raises("ascii", language.parse_response, {}, "foo:b\xf0")
tutils.raises("ascii", language.parse_response, "foo:b\xf0")
def test_parse_header(self):
r = language.parse_response({}, '400:h"foo"="bar"')
r = language.parse_response('400:h"foo"="bar"')
assert utils.get_header("foo", r.headers)
def test_parse_pause_before(self):
r = language.parse_response({}, "400:p0,10")
r = language.parse_response("400:p0,10")
assert r.actions[0].spec() == "p0,10"
def test_parse_pause_after(self):
r = language.parse_response({}, "400:pa,10")
r = language.parse_response("400:pa,10")
assert r.actions[0].spec() == "pa,10"
def test_parse_pause_random(self):
r = language.parse_response({}, "400:pr,10")
r = language.parse_response("400:pr,10")
assert r.actions[0].spec() == "pr,10"
def test_parse_stress(self):
# While larger values are known to work on linux,
# len() technically returns an int and a python 2.7 int on windows has 32bit precision.
# Therefore, we should keep the body length < 2147483647 bytes in our tests.
r = language.parse_response({}, "400:b@1g")
# While larger values are known to work on linux, len() technically
# returns an int and a python 2.7 int on windows has 32bit precision.
# Therefore, we should keep the body length < 2147483647 bytes in our
# tests.
r = language.parse_response("400:b@1g")
assert r.length({})
def test_spec(self):
def rt(s):
s = language.parse_response({}, s).spec()
assert language.parse_response({}, s).spec() == s
s = language.parse_response(s).spec()
assert language.parse_response(s).spec() == s
rt("400:b@100g")
rt("400")
rt("400:da")
def test_read_file():
tutils.raises(language.FileAccessDenied, language.read_file, {}, "=/foo")
p = tutils.test_data.path("data")
d = dict(staticdir=p)
assert language.read_file(d, "+./file").strip() == "testfile"
assert language.read_file(d, "+file").strip() == "testfile"
tutils.raises(language.FileAccessDenied, language.read_file, d, "+./nonexistent")
tutils.raises(language.FileAccessDenied, language.read_file, d, "+/nonexistent")
tutils.raises(language.FileAccessDenied, language.read_file, d, "+../test_language.py")
tutils.raises(
language.FileAccessDenied,
language.read_file,
d,
"+./nonexistent"
)
tutils.raises(
language.FileAccessDenied,
language.read_file,
d,
"+/nonexistent"
)
tutils.raises(
language.FileAccessDenied,
language.read_file,
d,
"+../test_language.py"
)
d["unconstrained_file_access"] = True
assert language.read_file(d, "+../test_language.py")

45
test/test_pathoc.py
View File

@ -1,6 +1,8 @@
import json
import cStringIO
from libpathod import pathoc, test, version, pathod
import re
from libpathod import pathoc, test, version, pathod, language
import tutils
@ -18,7 +20,9 @@ class _TestDaemon:
ssl=self.ssl,
ssloptions=self.ssloptions,
staticdir=tutils.test_data.path("data"),
anchors=[("/anchor/.*", "202")]
anchors=[
(re.compile("/anchor/.*"), language.parse_response("202"))
]
)
@classmethod
@ -37,17 +41,29 @@ class _TestDaemon:
r = c.request("get:/api/info")
assert tuple(json.loads(r.content)["version"]) == version.IVERSION
def tval(self, requests, showreq=False, showresp=False, explain=False,
showssl=False, hexdump=False, timeout=None, ignorecodes=None,
ignoretimeout=None):
def tval(
self,
requests,
showreq=False,
showresp=False,
explain=False,
showssl=False,
hexdump=False,
timeout=None,
ignorecodes=None,
ignoretimeout=None
):
c = pathoc.Pathoc(("127.0.0.1", self.d.port), ssl=self.ssl)
c.connect()
if timeout:
c.settimeout(timeout)
s = cStringIO.StringIO()
for i in requests:
r = language.parse_requests(i)[0]
if explain:
r = r.freeze({})
c.print_request(
i,
r,
showreq = showreq,
showresp = showresp,
explain = explain,
@ -125,25 +141,17 @@ class TestDaemon(_TestDaemon):
assert "HTTP/" in v
def test_explain(self):
reqs = [ "get:/p/200:b@100" ]
assert not "b@100" in self.tval(reqs, explain=True)
reqs = ["get:/p/200:b@100"]
assert "b@100" not in self.tval(reqs, explain=True)
def test_showreq(self):
reqs = [ "get:/api/info:p0,0", "get:/api/info:p0,0" ]
reqs = ["get:/api/info:p0,0", "get:/api/info:p0,0"]
assert self.tval(reqs, showreq=True).count("unprintables escaped") == 2
assert self.tval(reqs, showreq=True, hexdump=True).count("hex dump") == 2
def test_parse_err(self):
assert "Error parsing" in self.tval(["foo"])
def test_conn_err(self):
assert "Invalid server response" in self.tval(["get:'/p/200:d2'"])
def test_fileread(self):
d = tutils.test_data.path("data/request")
assert "foo" in self.tval(["+%s"%d], showreq=True)
assert "File" in self.tval(["+/nonexistent"])
def test_connect_fail(self):
to = ("foobar", 80)
c = pathoc.Pathoc(("127.0.0.1", self.d.port))
@ -157,6 +165,3 @@ class TestDaemon(_TestDaemon):
"HTTP/1.1 200 OK\r\n"
)
c.http_connect(to)

35
test/test_pathod.py
View File

@ -1,19 +1,9 @@
import pprint
from libpathod import pathod, version
from netlib import tcp, http, certutils
import requests
from netlib import tcp, http
import tutils
class TestPathod:
def test_instantiation(self):
p = pathod.Pathod(
("127.0.0.1", 0),
anchors = [(".*", "200:da")]
)
assert p.anchors
tutils.raises("invalid regex", pathod.Pathod, ("127.0.0.1", 0), anchors=[("*", "200:da")])
tutils.raises("invalid page spec", pathod.Pathod, ("127.0.0.1", 0), anchors=[("foo", "bar")])
class TestPathod:
def test_logging(self):
p = pathod.Pathod(("127.0.0.1", 0))
assert len(p.get_log()) == 0
@ -30,6 +20,7 @@ class TestPathod:
class TestNoWeb(tutils.DaemonTests):
noweb = True
def test_noweb(self):
assert self.get("200:da").status_code == 200
assert self.getpath("/").status_code == 800
@ -37,6 +28,7 @@ class TestNoWeb(tutils.DaemonTests):
class TestTimeout(tutils.DaemonTests):
timeout = 0.01
def test_noweb(self):
# FIXME: Add float values to spec language, reduce test timeout to
# increase test performance
@ -46,6 +38,7 @@ class TestTimeout(tutils.DaemonTests):
class TestNoApi(tutils.DaemonTests):
noapi = True
def test_noapi(self):
assert self.getpath("/log").status_code == 404
r = self.getpath("/")
@ -59,7 +52,10 @@ class TestNotAfterConnect(tutils.DaemonTests):
not_after_connect = True
)
def test_connect(self):
r = self.pathoc(r"get:'http://foo.com/p/202':da", connect_to=("localhost", self.d.port))
r = self.pathoc(
r"get:'http://foo.com/p/202':da",
connect_to=("localhost", self.d.port)
)
assert r.status_code == 202
@ -157,11 +153,15 @@ class CommonTests(tutils.DaemonTests):
assert l["type"] == "error"
assert "foo" in l["msg"]
def test_invalid_body(self):
tutils.raises(http.HttpError, self.pathoc, "get:/:h'content-length'='foo'")
def test_invalid_content_length(self):
tutils.raises(
http.HttpError,
self.pathoc,
"get:/:h'content-length'='foo'"
)
l = self.d.last_log()
assert l["type"] == "error"
assert "Invalid" in l["msg"]
assert "Content-Length unknown" in l["msg"]
def test_invalid_headers(self):
tutils.raises(http.HttpError, self.pathoc, "get:/:h'\t'='foo'")
@ -204,7 +204,7 @@ class TestDaemon(CommonTests):
class TestDaemonSSL(CommonTests):
ssl = True
def test_ssl_conn_failure(self):
def _test_ssl_conn_failure(self):
c = tcp.TCPClient(("localhost", self.d.port))
c.rbufsize = 0
c.wbufsize = 0
@ -222,4 +222,3 @@ class TestDaemonSSL(CommonTests):
r = self.pathoc(r"get:/p/202")
assert r.status_code == 202
assert self.d.last_log()["cipher"][1] > 0

20
test/test_test.py
View File

@ -4,13 +4,18 @@ from libpathod import test
import tutils
logging.disable(logging.CRITICAL)
class TestDaemonManual:
def test_simple(self):
with test.Daemon() as d:
rsp = requests.get("http://localhost:%s/p/202:da"%d.port)
assert rsp.ok
assert rsp.status_code == 202
tutils.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202:da"%d.port)
tutils.raises(
"Connection aborted",
requests.get,
"http://localhost:%s/p/202:da"%d.port
)
def test_startstop_ssl(self):
d = test.Daemon(ssl=True)
@ -18,7 +23,11 @@ class TestDaemonManual:
assert rsp.ok
assert rsp.status_code == 202
d.shutdown()
tutils.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202:da"%d.port)
tutils.raises(
"Connection aborted",
requests.get,
"http://localhost:%s/p/202:da"%d.port
)
def test_startstop_ssl_explicit(self):
ssloptions = dict(
@ -31,6 +40,9 @@ class TestDaemonManual:
assert rsp.ok
assert rsp.status_code == 202
d.shutdown()
tutils.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202:da"%d.port)
tutils.raises(
"Connection aborted",
requests.get,
"http://localhost:%s/p/202:da"%d.port
)

21
test/tutils.py
View File

@ -1,8 +1,12 @@
import tempfile, os, shutil
import tempfile
import os
import re
import shutil
from contextlib import contextmanager
from libpathod import utils, test, pathoc, pathod
from libpathod import utils, test, pathoc, pathod, language
import requests
class DaemonTests:
noweb = False
noapi = False
@ -11,6 +15,7 @@ class DaemonTests:
timeout = None
hexdump = False
ssloptions = None
@classmethod
def setUpAll(self):
opts = self.ssloptions or {}
@ -19,7 +24,9 @@ class DaemonTests:
so = pathod.SSLOptions(**opts)
self.d = test.Daemon(
staticdir=test_data.path("data"),
anchors=[("/anchor/.*", "202:da")],
anchors=[
(re.compile("/anchor/.*"), language.parse_response("202:da"))
],
ssl = self.ssl,
ssloptions = so,
sizelimit=1*1024*1024,
@ -45,7 +52,13 @@ class DaemonTests:
def getpath(self, path, params=None):
scheme = "https" if self.ssl else "http"
return requests.get(
"%s://localhost:%s/%s"%(scheme, self.d.port, path), verify=False, params=params
"%s://localhost:%s/%s"%(
scheme,
self.d.port,
path
),
verify=False,
params=params
)
def get(self, spec):