什么是twisted？
twisted是一个用python语言写的事件驱动的网络框架，他支持很多种协议，包括UDP,TCP,TLS和其他应用层协议，比如HTTP，SMTP，NNTM，IRC，XMPP/Jabber。 非常好的一点是twisted实现和很多应用层的协议，开发人员可以直接只用这些协议的实现。其实要修改Twisted的SSH服务器端实现非常简单。很多时候，开发人员需要实现protocol类。
一个Twisted程序由reactor发起的主循环和一些回调函数组成。当事件发生了，比如一个client连接到了server，这时候服务器端的事件会被触发执行。
用Twisted写一个简单的TCP服务器
下面的代码是一个TCPServer，这个server记录客户端发来的数据信息。
|
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
|
==== code1.py ====
import sys
from twisted.internet.protocol import ServerFactory
from twisted.protocols.basic import LineReceiver
from twisted.python import log
from twisted.internet import reactor
class CmdProtocol(LineReceiver):
delimiter = '\n'
def connectionMade(self):
self.client_ip = self.transport.getPeer()[1]
log.msg("Client connection from %s" % self.client_ip)
if len(self.factory.clients) >= self.factory.clients_max:
log.msg("Too many connections. bye !")
self.client_ip = None
self.transport.loseConnection()
else:
self.factory.clients.append(self.client_ip)
def connectionLost(self, reason):
log.msg('Lost client connection. Reason: %s' % reason)
if self.client_ip:
self.factory.clients.remove(self.client_ip)
def lineReceived(self, line):
log.msg('Cmd received from %s : %s' % (self.client_ip, line))
class MyFactory(ServerFactory):
protocol = CmdProtocol
def __init__(self, clients_max=10):
self.clients_max = clients_max
self.clients = []
log.startLogging(sys.stdout)
reactor.listenTCP(9999, MyFactory(2))
reactor.run()
|
下面的代码至关重要：
|
1
2
|
from twisted.internet import reactor
reactor.run()
|
这两行代码会启动reator的主循环。
在上面的代码中我们创建了"ServerFactory"类，这个工厂类负责返回“CmdProtocol”的实例。 每一个连接都由实例化的“CmdProtocol”实例来做处理。 Twisted的reactor会在TCP连接上后自动创建CmdProtocol的实例。如你所见，protocol类的方法都对应着一种事件处理。
当client连上server之后会触发“connectionMade"方法，在这个方法中你可以做一些鉴权之类的操作，也可以限制客户端的连接总数。每一个protocol的实例都有一个工厂的引用，使用self.factory可以访问所在的工厂实例。
上面实现的”CmdProtocol“是twisted.protocols.basic.LineReceiver的子类，LineReceiver类会将客户端发送的数据按照换行符分隔，每到一个换行符都会触发lineReceived方法。稍后我们可以增强LineReceived来解析命令。
Twisted实现了自己的日志系统，这里我们配置将日志输出到stdout
当执行reactor.listenTCP时我们将工厂绑定到了9999端口开始监听。
|
1
2
3
4
5
6
|
user@lab:~/TMP$ python code1.py
2011-08-29 13:32:32+0200 [-] Log opened.
2011-08-29 13:32:32+0200 [-] __main__.MyFactory starting on 9999
2011-08-29 13:32:32+0200 [-] Starting factory <__main__.MyFactory instance at 0x227e320
2011-08-29 13:32:35+0200 [__main__.MyFactory] Client connection from 127.0.0.1
2011-08-29 13:32:38+0200 [CmdProtocol,0,127.0.0.1] Cmd received from 127.0.0.1 : hello server
|
使用Twisted来调用外部进程
下面我们给前面的server添加一个命令，通过这个命令可以读取/var/log/syslog的内容
|
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
|
import sys
import os
from twisted.internet.protocol import ServerFactory, ProcessProtocol
from twisted.protocols.basic import LineReceiver
from twisted.python import log
from twisted.internet import reactor
class TailProtocol(ProcessProtocol):
def __init__(self, write_callback):
self.write = write_callback
def outReceived(self, data):
self.write("Begin lastlog\n")
data = [line for line in data.split('\n') if not line.startswith('==')]
for d in data:
self.write(d + '\n')
self.write("End lastlog\n")
def processEnded(self, reason):
if reason.value.exitCode != 0:
log.msg(reason)
class CmdProtocol(LineReceiver):
delimiter = '\n'
def processCmd(self, line):
if line.startswith('lastlog'):
tailProtocol = TailProtocol(self.transport.write)
reactor.spawnProcess(tailProtocol, '/usr/bin/tail', args=['/usr/bin/tail', '-10', '/var/log/syslog'])
elif line.startswith('exit'):
self.transport.loseConnection()
else:
self.transport.write('Command not found.\n')
def connectionMade(self):
self.client_ip = self.transport.getPeer()[1]
log.msg("Client connection from %s" % self.client_ip)
if len(self.factory.clients) >= self.factory.clients_max:
log.msg("Too many connections. bye !")
self.client_ip = None
self.transport.loseConnection()
else:
self.factory.clients.append(self.client_ip)
def connectionLost(self, reason):
log.msg('Lost client connection. Reason: %s' % reason)
if self.client_ip:
self.factory.clients.remove(self.client_ip)
def lineReceived(self, line):
log.msg('Cmd received from %s : %s' % (self.client_ip, line))
self.processCmd(line)
class MyFactory(ServerFactory):
protocol = CmdProtocol
def __init__(self, clients_max=10):
self.clients_max = clients_max
self.clients = []
log.startLogging(sys.stdout)
reactor.listenTCP(9999, MyFactory(2))
reactor.run()
|
在上面的代码中，没从客户端接收到一行内容后会执行processCmd方法，如果收到的一行内容是exit命令，那么服务器端会断开连接，如果收到的是lastlog，我们要吐出一个子进程来执行tail命令，并将tail命令的输出重定向到客户端。这里我们需要实现ProcessProtocol类，需要重写该类的processEnded方法和outReceived方法。在tail命令有输出时会执行outReceived方法，当进程退出时会执行processEnded方法。
如下是执行结果样例：
|
1
2
3
4
5
6
7
8
9
|
user@lab:~/TMP$ python code2.py
2011-08-29 15:13:38+0200 [-] Log opened.
2011-08-29 15:13:38+0200 [-] __main__.MyFactory starting on 9999
2011-08-29 15:13:38+0200 [-] Starting factory <__main__.MyFactory instance at 0x1a5a3f8>
2011-08-29 15:13:47+0200 [__main__.MyFactory] Client connection from 127.0.0.1
2011-08-29 15:13:58+0200 [CmdProtocol,0,127.0.0.1] Cmd received from 127.0.0.1 : test
2011-08-29 15:14:02+0200 [CmdProtocol,0,127.0.0.1] Cmd received from 127.0.0.1 : lastlog
2011-08-29 15:14:05+0200 [CmdProtocol,0,127.0.0.1] Cmd received from 127.0.0.1 : exit
2011-08-29 15:14:05+0200 [CmdProtocol,0,127.0.0.1] Lost client connection. Reason: [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.error.ConnectionDone'>: Connection was closed cleanly.
|
可以使用下面的命令作为客户端发起命令：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
user@lab:~$ netcat 127.0.0.1 9999
test
Command not found.
lastlog
Begin lastlog
Aug 29 15:02:03 lab sSMTP[5919]: Unable to locate mail
Aug 29 15:02:03 lab sSMTP[5919]: Cannot open mail:25
Aug 29 15:02:03 lab CRON[4945]: (CRON) error (grandchild #4947 failed with exit status 1)
Aug 29 15:02:03 lab sSMTP[5922]: Unable to locate mail
Aug 29 15:02:03 lab sSMTP[5922]: Cannot open mail:25
Aug 29 15:02:03 lab CRON[4945]: (logcheck) MAIL (mailed 1 byte of output; but got status 0x0001, #012)
Aug 29 15:05:01 lab CRON[5925]: (root) CMD (command -v debian-sa1 > /dev/null && debian-sa1 1 1)
Aug 29 15:10:01 lab CRON[5930]: (root) CMD (test -x /usr/lib/atsar/atsa1 && /usr/lib/atsar/atsa1)
Aug 29 15:10:01 lab CRON[5928]: (CRON) error (grandchild #5930 failed with exit status 1)
Aug 29 15:13:21 lab pulseaudio[3361]: ratelimit.c: 387 events suppressed
End lastlog
exit
|
使用Deferred对象
reactor是一个循环，这个循环在等待事件的发生。 这里的事件可以是数据库操作，也可以是长时间的计算操作。 只要这些操作可以返回一个Deferred对象。Deferred对象可以自动得在事件发生时触发回调函数。reactor会block当前代码的执行。
现在我们要使用Defferred对象来计算SHA1哈希。
|
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
|
import sys
import os
import hashlib
from twisted.internet.protocol import ServerFactory, ProcessProtocol
from twisted.protocols.basic import LineReceiver
from twisted.python import log
from twisted.internet import reactor, threads
class TailProtocol(ProcessProtocol):
def __init__(self, write_callback):
self.write = write_callback
def outReceived(self, data):
self.write("Begin lastlog\n")
data = [line for line in data.split('\n') if not line.startswith('==')]
for d in data:
self.write(d + '\n')
self.write("End lastlog\n")
def processEnded(self, reason):
if reason.value.exitCode != 0:
log.msg(reason)
class HashCompute(object):
def __init__(self, path, write_callback):
self.path = path
self.write = write_callback
def blockingMethod(self):
os.path.isfile(self.path)
data = file(self.path).read()
# uncomment to add more delay
# import time
# time.sleep(10)
return hashlib.sha1(data).hexdigest()
def compute(self):
d = threads.deferToThread(self.blockingMethod)
d.addCallback(self.ret)
d.addErrback(self.err)
def ret(self, hdata):
self.write("File hash is : %s\n" % hdata)
def err(self, failure):
self.write("An error occured : %s\n" % failure.getErrorMessage())
class CmdProtocol(LineReceiver):
delimiter = '\n'
def processCmd(self, line):
if line.startswith('lastlog'):
tailProtocol = TailProtocol(self.transport.write)
reactor.spawnProcess(tailProtocol, '/usr/bin/tail', args=['/usr/bin/tail', '-10', '/var/log/syslog'])
elif line.startswith('comphash'):
try:
useless, path = line.split(' ')
except:
self.transport.write('Please provide a path.\n')
return
hc = HashCompute(path, self.transport.write)
hc.compute()
elif line.startswith('exit'):
self.transport.loseConnection()
else:
self.transport.write('Command not found.\n')
def connectionMade(self):
self.client_ip = self.transport.getPeer()[1]
log.msg("Client connection from %s" % self.client_ip)
if len(self.factory.clients) >= self.factory.clients_max:
log.msg("Too many connections. bye !")
self.client_ip = None
self.transport.loseConnection()
else:
self.factory.clients.append(self.client_ip)
def connectionLost(self, reason):
log.msg('Lost client connection. Reason: %s' % reason)
if self.client_ip:
self.factory.clients.remove(self.client_ip)
def lineReceived(self, line):
log.msg('Cmd received from %s : %s' % (self.client_ip, line))
self.processCmd(line)
class MyFactory(ServerFactory):
protocol = CmdProtocol
def __init__(self, clients_max=10):
self.clients_max = clients_max
self.clients = []
log.startLogging(sys.stdout)
reactor.listenTCP(9999, MyFactory(2))
reactor.run()
|
blockingMethod从文件系统读取一个文件计算SHA1，这里我们使用twisted的deferToThread方法，这个方法返回一个Deferred对象。这里的Deferred对象是调用后马上就返回了，这样主进程就可以继续执行处理其他的事件。当传给deferToThread的方法执行完毕后会马上触发其回调函数。如果执行中出错，blockingMethod方法会抛出异常。如果成功执行会通过hdata的ret返回计算的结果。
推荐的twisted阅读资料
http://twistedmatrix.com/documents/current/core/howto/defer.html http://twistedmatrix.com/documents/current/core/howto/process.html http://twistedmatrix.com/documents/current/core/howto/servers.html
API文档：
http://twistedmatrix.com/documents/current/api/twisted.html