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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
|
#!/usr/bin/env python
"""
SSL server for Forge tests.
- The server changes to the directory of the server script.
- SSL uses "server.key" and "server.crt".
- Sever performs basic static file serving.
- Starts Flash cross domain policy file server.
- Defaults to HTTP/HTTPS port 19400.
- Defaults to Flash socket policy port 19945.
$ ./server.py [options]
If you just need a simple HTTP server, also consider:
$ python -m SimpleHTTPServer 19400
"""
from multiprocessing import Process
from optparse import OptionParser
import SimpleHTTPServer
import SocketServer
import os
import sys
import time
# Try to import special Forge SSL module with session cache support
# Using the built directory directly
python_version = "python" + sys.version[:3]
sys.path.insert(0, os.path.join(
os.path.dirname(os.path.realpath(__file__)),
"..", "dist", "forge_ssl", "lib", python_version, "site-packages"))
try:
from forge import ssl
have_ssl_sessions = True
have_ssl = True
except ImportError:
have_ssl_sessions = False
try:
import ssl
have_ssl = True
except ImportError:
have_ssl = False
# Set address reuse for all TCPServers
SocketServer.TCPServer.allow_reuse_address = True
# The policy file
# NOTE: This format is very strict. Edit with care.
policy_file = """\
<?xml version="1.0"?>\
<!DOCTYPE cross-domain-policy\
SYSTEM "http://www.adobe.com/xml/dtds/cross-domain-policy.dtd">\
<cross-domain-policy>\
<allow-access-from domain="*" to-ports="*"/>\
</cross-domain-policy>\0"""
class PolicyHandler(SocketServer.BaseRequestHandler):
"""
The RequestHandler class for our server.
Returns a policy file when requested.
"""
def handle(self):
# get some data
# TODO: make this more robust (while loop, etc)
self.data = self.request.recv(1024).rstrip('\0')
#print "%s wrote:" % self.client_address[0]
#print repr(self.data)
# if policy file request, send the file.
if self.data == "<policy-file-request/>":
print "Policy server request from %s." % (self.client_address[0])
self.request.send(policy_file)
else:
print "Policy server received junk from %s: \"%s\"" % \
(self.client_address[0], repr(self.data))
def create_policy_server(options):
"""Start a policy server"""
print "Policy serving from %d." % (options.policy_port)
policyd = SocketServer.TCPServer((options.host, options.policy_port), PolicyHandler)
return policyd
class ThreadedTCPServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
pass
def create_http_server(options, script_dir):
"""Start a static file server"""
# use UTF-8 encoding for javascript files
m = SimpleHTTPServer.SimpleHTTPRequestHandler.extensions_map
m['.js'] = 'application/javascript;charset=UTF-8'
Handler = SimpleHTTPServer.SimpleHTTPRequestHandler
# httpd = SocketServer.TCPServer((options.host, options.port), Handler)
httpd = ThreadedTCPServer((options.host, options.port), Handler)
if options.tls:
if not have_ssl:
raise Exception("SSL support from Python 2.7 or later is required.")
# setup session args if we session support
sess_args = {}
if have_ssl_sessions:
sess_args["sess_id_ctx"] = "forgetest"
else:
print "Forge SSL with session cache not available, using standard version."
httpd.socket = ssl.wrap_socket(
httpd.socket,
keyfile="server.key",
certfile="server.crt",
server_side=True,
**sess_args)
print "Serving from \"%s\"." % (script_dir)
print "%s://%s:%d/" % \
(("https" if options.tls else "http"),
httpd.server_address[0],
options.port)
return httpd
def serve_forever(server):
"""Serve until shutdown or keyboard interrupt."""
try:
server.serve_forever()
except KeyboardInterrupt:
return
def main():
"""Start static file and policy servers"""
usage = "Usage: %prog [options]"
parser = OptionParser(usage=usage)
parser.add_option("", "--host", dest="host", metavar="HOST",
default="localhost", help="bind to HOST")
parser.add_option("-p", "--port", dest="port", type="int",
help="serve on PORT", metavar="PORT", default=19400)
parser.add_option("-P", "--policy-port", dest="policy_port", type="int",
help="serve policy file on PORT", metavar="PORT", default=19945)
parser.add_option("", "--tls", dest="tls", action="store_true",
help="serve HTTPS", default=False)
(options, args) = parser.parse_args()
# Change to script dir so SSL and test files are in current dir.
script_dir = os.path.dirname(os.path.realpath(__file__))
os.chdir(script_dir)
print "Forge Test Server. Use ctrl-c to exit."
# create policy and http servers
httpd = create_http_server(options, script_dir)
policyd = create_policy_server(options)
# start servers
server_p = Process(target=serve_forever, args=(httpd,))
policy_p = Process(target=serve_forever, args=(policyd,))
server_p.start()
policy_p.start()
processes = [server_p, policy_p]
while len(processes) > 0:
try:
for p in processes:
if p.is_alive():
p.join(1)
else:
processes.remove(p)
except KeyboardInterrupt:
print "\nStopping test server..."
# processes each receive interrupt
# so no need to shutdown
#httpd.shutdown();
#policyd.shutdown();
if __name__ == "__main__":
main()
|
