1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
|
import netlib.websockets
import pyparsing as pp
from . import base, generators, actions, message
"""
wf:ctext:b'foo'
wf:c15:r'foo'
wf:fin:rsv1:rsv2:rsv3:mask
wf:-fin:-rsv1:-rsv2:-rsv3:-mask
wf:p234
wf:m"mask"
"""
class WF(base.CaselessLiteral):
TOK = "wf"
class Code(base.IntField):
names = {
"continue": netlib.websockets.OPCODE.CONTINUE,
"text": netlib.websockets.OPCODE.TEXT,
"binary": netlib.websockets.OPCODE.BINARY,
"close": netlib.websockets.OPCODE.CLOSE,
"ping": netlib.websockets.OPCODE.PING,
"pong": netlib.websockets.OPCODE.PONG,
}
max = 15
preamble = "c"
class Body(base.Value):
preamble = "b"
class WebsocketFrame(message.Message):
comps = (
Body,
Code,
actions.PauseAt,
actions.DisconnectAt,
actions.InjectAt
)
logattrs = ["body"]
@property
def actions(self):
return self.toks(actions._Action)
@property
def body(self):
return self.tok(Body)
@property
def code(self):
return self.tok(Code)
@classmethod
def expr(klass):
parts = [i.expr() for i in klass.comps]
atom = pp.MatchFirst(parts)
resp = pp.And(
[
WF.expr(),
base.Sep,
pp.ZeroOrMore(base.Sep + atom)
]
)
resp = resp.setParseAction(klass)
return resp
def values(self, settings):
vals = []
if self.body:
bodygen = self.body.value.get_generator(settings)
length = len(self.body.value.get_generator(settings))
else:
bodygen = None
length = 0
frameparts = dict(
mask = True,
payload_length = length
)
if self.code:
frameparts["opcode"] = self.code.value
frame = netlib.websockets.FrameHeader(**frameparts)
vals = [frame.to_bytes()]
if self.body:
masker = netlib.websockets.Masker(frame.masking_key)
vals.append(
generators.TransformGenerator(
bodygen,
masker.mask
)
)
return vals
def resolve(self, settings, msg=None):
return self.__class__(
[i.resolve(settings, msg) for i in self.tokens]
)
def spec(self):
return ":".join([i.spec() for i in self.tokens])
|
