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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
# -*- test-case-name: twisted.test.test_postfix -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Postfix mail transport agent related protocols.
"""
import sys
try:
# Python 2
from UserDict import UserDict
except ImportError:
# Python 3
from collections import UserDict
try:
# Python 2
from urllib import quote as _quote, unquote as _unquote
except ImportError:
# Python 3
from urllib.parse import quote as _quote, unquote as _unquote
from twisted.protocols import basic
from twisted.protocols import policies
from twisted.internet import protocol, defer
from twisted.python import log
from twisted.python.compat import unicode
# urllib's quote functions just happen to match
# the postfix semantics.
def quote(s):
quoted = _quote(s)
if isinstance(quoted, unicode):
quoted = quoted.encode("ascii")
return quoted
def unquote(s):
if isinstance(s, bytes):
s = s.decode("ascii")
quoted = _unquote(s)
return quoted.encode("ascii")
class PostfixTCPMapServer(basic.LineReceiver, policies.TimeoutMixin):
"""
Postfix mail transport agent TCP map protocol implementation.
Receive requests for data matching given key via lineReceived,
asks it's factory for the data with self.factory.get(key), and
returns the data to the requester. None means no entry found.
You can use postfix's postmap to test the map service::
/usr/sbin/postmap -q KEY tcp:localhost:4242
"""
timeout = 600
delimiter = b'\n'
def connectionMade(self):
self.setTimeout(self.timeout)
def sendCode(self, code, message=b''):
"""
Send an SMTP-like code with a message.
"""
self.sendLine(str(code).encode("ascii") + b' ' + message)
def lineReceived(self, line):
self.resetTimeout()
try:
request, params = line.split(None, 1)
except ValueError:
request = line
params = None
try:
f = getattr(self, u'do_' + request.decode("ascii"))
except AttributeError:
self.sendCode(400, b'unknown command')
else:
try:
f(params)
except:
excInfo = str(sys.exc_info()[1]).encode("ascii")
self.sendCode(400, b'Command ' + request + b' failed: ' +
excInfo)
def do_get(self, key):
if key is None:
self.sendCode(400, b"Command 'get' takes 1 parameters.")
else:
d = defer.maybeDeferred(self.factory.get, key)
d.addCallbacks(self._cbGot, self._cbNot)
d.addErrback(log.err)
def _cbNot(self, fail):
msg = fail.getErrorMessage().encode("ascii")
self.sendCode(400, msg)
def _cbGot(self, value):
if value is None:
self.sendCode(500)
else:
self.sendCode(200, quote(value))
def do_put(self, keyAndValue):
if keyAndValue is None:
self.sendCode(400, b"Command 'put' takes 2 parameters.")
else:
try:
key, value = keyAndValue.split(None, 1)
except ValueError:
self.sendCode(400, b"Command 'put' takes 2 parameters.")
else:
self.sendCode(500, b'put is not implemented yet.')
class PostfixTCPMapDictServerFactory(UserDict, protocol.ServerFactory):
"""
An in-memory dictionary factory for PostfixTCPMapServer.
"""
protocol = PostfixTCPMapServer
class PostfixTCPMapDeferringDictServerFactory(protocol.ServerFactory):
"""
An in-memory dictionary factory for PostfixTCPMapServer.
"""
protocol = PostfixTCPMapServer
def __init__(self, data=None):
self.data = {}
if data is not None:
self.data.update(data)
def get(self, key):
return defer.succeed(self.data.get(key))
|
