Fri, 22 Aug 2008 21:09:04 +0000
First commit, where do you want to go tomorrow?
0 | 1 | require "luarocks.require" |
2 | ||
3 | require "copas" | |
4 | require "socket" | |
5 | require "ssl" | |
6 | require "lxp" | |
7 | ||
8 | function log(type, area, message) | |
9 | print(type, area, message); | |
10 | end | |
11 | ||
12 | require "core.stanza_dispatch" | |
13 | require "core.rostermanager" | |
14 | require "core.offlinemessage" | |
15 | require "util.stanza" | |
16 | require "util.jid" | |
17 | ||
18 | -- Locals for faster access -- | |
19 | local t_insert = table.insert; | |
20 | local t_concat = table.concat; | |
21 | local t_concatall = function (t, sep) local tt = {}; for _, s in ipairs(t) do t_insert(tt, tostring(s)); end return t_concat(tt, sep); end | |
22 | local m_random = math.random; | |
23 | local format = string.format; | |
24 | local st = stanza; | |
25 | ------------------------------ | |
26 | ||
27 | users = {}; | |
28 | hosts = { | |
29 | ["localhost"] = { | |
30 | type = "local"; | |
31 | connected = true; | |
32 | sessions = {}; | |
33 | }; | |
34 | ["getjabber.ath.cx"] = { | |
35 | type = "local"; | |
36 | connected = true; | |
37 | sessions = {}; | |
38 | }; | |
39 | } | |
40 | ||
41 | local hosts, users = hosts, users; | |
42 | ||
43 | local ssl_ctx, msg = ssl.newcontext { mode = "server", protocol = "sslv23", key = "/home/matthew/ssl_cert/server.key", | |
44 | certificate = "/home/matthew/ssl_cert/server.crt", capath = "/etc/ssl/certs", verify = "none", } | |
45 | ||
46 | if not ssl_ctx then error("Failed to initialise SSL/TLS support: "..tostring(msg)); end | |
47 | ||
48 | ||
49 | function connect_host(host) | |
50 | hosts[host] = { type = "remote", sendbuffer = {} }; | |
51 | end | |
52 | ||
53 | function handler(conn) | |
54 | local copas_receive, copas_send = copas.receive, copas.send; | |
55 | local reqdata, sktmsg; | |
56 | local session = { sendbuffer = { external = {} }, conn = conn, notopen = true, priority = 0 } | |
57 | ||
58 | ||
59 | -- Logging functions -- | |
60 | ||
61 | local mainlog, log = log; | |
62 | do | |
63 | local conn_name = tostring(conn):match("%w+$"); | |
64 | log = function (type, area, message) mainlog(type, conn_name, message); end | |
65 | end | |
66 | local print = function (...) log("info", "core", t_concatall({...}, "\t")); end | |
67 | session.log = log; | |
68 | ||
69 | -- -- -- | |
70 | ||
71 | -- Send buffers -- | |
72 | ||
73 | local sendbuffer = session.sendbuffer; | |
74 | local send = function (data) return t_insert(sendbuffer, tostring(data)); end; | |
75 | local send_to = function (to, stanza) | |
76 | local node, host, resource = jid.split(to); | |
77 | print("Routing stanza to "..to..":", node, host, resource); | |
78 | if not hosts[host] then | |
79 | print(" ...but host offline, establishing connection"); | |
80 | connect_host(host); | |
81 | t_insert(hosts[host].sendbuffer, stanza); -- This will be sent when s2s connection succeeds | |
82 | elseif hosts[host].connected then | |
83 | print(" ...putting in our external send buffer"); | |
84 | t_insert(sendbuffer.external, { node = node, host = host, resource = resource, data = stanza}); | |
85 | print(" ...there are now "..tostring(#sendbuffer.external).." stanzas in the external send buffer"); | |
86 | end | |
87 | end | |
88 | session.send, session.send_to = send, send_to; | |
89 | ||
90 | -- -- -- | |
91 | print("Client connected"); | |
92 | conn = ssl.wrap(copas.wrap(conn), ssl_ctx); | |
93 | ||
94 | do | |
95 | local succ, msg | |
96 | conn:settimeout(15) | |
97 | while not succ do | |
98 | succ, msg = conn:dohandshake() | |
99 | if not succ then | |
100 | print("SSL: "..tostring(msg)); | |
101 | if msg == 'wantread' then | |
102 | socket.select({conn}, nil) | |
103 | elseif msg == 'wantwrite' then | |
104 | socket.select(nil, {conn}) | |
105 | else | |
106 | -- other error | |
107 | end | |
108 | end | |
109 | end | |
110 | end | |
111 | print("SSL handshake complete"); | |
112 | -- XML parser initialisation -- | |
113 | ||
114 | local parser; | |
115 | local stanza; | |
116 | ||
117 | local stanza_dispatch = init_stanza_dispatcher(session); | |
118 | ||
119 | local xml_handlers = {}; | |
120 | ||
121 | do | |
122 | local ns_stack = { "" }; | |
123 | local curr_ns = ""; | |
124 | local curr_tag; | |
125 | function xml_handlers:StartElement(name, attr) | |
126 | curr_ns,name = name:match("^(.+):(%w+)$"); | |
127 | print("Tag received:", name, tostring(curr_ns)); | |
128 | if not stanza then | |
129 | if session.notopen then | |
130 | if name == "stream" then | |
131 | session.host = attr.to or error("Client failed to specify destination hostname"); | |
132 | session.version = attr.version or 0; | |
133 | session.streamid = m_random(1000000, 99999999); | |
134 | print(session, session.host, "Client opened stream"); | |
135 | send("<?xml version='1.0'?>"); | |
136 | send(format("<stream:stream xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams' id='%s' from='%s' >", session.streamid, session.host)); | |
137 | --send("<stream:features>"); | |
138 | --send("<mechanism>PLAIN</mechanism>"); | |
139 | --send [[<register xmlns="http://jabber.org/features/iq-register"/> ]] | |
140 | --send("</stream:features>"); | |
141 | log("info", "core", "Stream opened successfully"); | |
142 | session.notopen = nil; | |
143 | return; | |
144 | end | |
145 | error("Client failed to open stream successfully"); | |
146 | end | |
147 | if name ~= "iq" and name ~= "presence" and name ~= "message" then | |
148 | error("Client sent invalid top-level stanza"); | |
149 | end | |
150 | stanza = st.stanza(name, { to = attr.to, type = attr.type, id = attr.id, xmlns = curr_ns }); | |
151 | curr_tag = stanza; | |
152 | else | |
153 | attr.xmlns = curr_ns; | |
154 | stanza:tag(name, attr); | |
155 | end | |
156 | end | |
157 | function xml_handlers:CharacterData(data) | |
158 | if data:match("%S") then | |
159 | stanza:text(data); | |
160 | end | |
161 | end | |
162 | function xml_handlers:EndElement(name) | |
163 | curr_ns,name = name:match("^(.+):(%w+)$"); | |
164 | --print("<"..name.."/>", tostring(stanza), tostring(#stanza.last_add < 1), tostring(stanza.last_add[#stanza.last_add].name)); | |
165 | if (not stanza) or #stanza.last_add < 0 or (#stanza.last_add > 0 and name ~= stanza.last_add[#stanza.last_add].name) then error("XML parse error in client stream"); end | |
166 | -- Complete stanza | |
167 | print(name, tostring(#stanza.last_add)); | |
168 | if #stanza.last_add == 0 then | |
169 | stanza_dispatch(stanza); | |
170 | stanza = nil; | |
171 | else | |
172 | stanza:up(); | |
173 | end | |
174 | end | |
175 | --[[ function xml_handlers:StartNamespaceDecl(namespace) | |
176 | table.insert(ns_stack, namespace); | |
177 | curr_ns = namespace; | |
178 | log("debug", "parser", "Entering namespace "..tostring(curr_ns)); | |
179 | end | |
180 | function xml_handlers:EndNamespaceDecl(namespace) | |
181 | table.remove(ns_stack); | |
182 | log("debug", "parser", "Leaving namespace "..tostring(curr_ns)); | |
183 | curr_ns = ns_stack[#ns_stack]; | |
184 | log("debug", "parser", "Entering namespace "..tostring(curr_ns)); | |
185 | end | |
186 | ]] | |
187 | end | |
188 | parser = lxp.new(xml_handlers, ":"); | |
189 | ||
190 | -- -- -- | |
191 | ||
192 | -- Main loop -- | |
193 | print "Receiving..." | |
194 | reqdata = copas_receive(conn, 1); | |
195 | print "Received" | |
196 | while reqdata do | |
197 | parser:parse(reqdata); | |
198 | if #sendbuffer.external > 0 then | |
199 | -- Stanzas queued to go to other places, from us | |
200 | -- ie. other local users, or remote hosts that weren't connected before | |
201 | print(#sendbuffer.external.." stanzas queued for other recipients, sending now..."); | |
202 | for n, packet in pairs(sendbuffer.external) do | |
203 | if not hosts[packet.host] then | |
204 | connect_host(packet.host); | |
205 | t_insert(hosts[packet.host].sendbuffer, packet.data); | |
206 | elseif hosts[packet.host].type == "local" then | |
207 | print(" ...is to a local user") | |
208 | local destuser = hosts[packet.host].sessions[packet.node]; | |
209 | if destuser and destuser.sessions then | |
210 | if not destuser.sessions[packet.resource] then | |
211 | local best_resource; | |
212 | for resource, session in pairs(destuser.sessions) do | |
213 | if not best_session then best_session = session; | |
214 | elseif session.priority >= best_session.priority and session.priority >= 0 then | |
215 | best_session = session; | |
216 | end | |
217 | end | |
218 | if not best_session then | |
219 | offlinemessage.new(packet.node, packet.host, packet.data); | |
220 | else | |
221 | print("resource '"..packet.resource.."' was not online, have chosen to send to '"..best_session.username.."@"..best_session.host.."/"..best_session.resource.."'"); | |
222 | packet.resource = best_session.resource; | |
223 | end | |
224 | end | |
225 | if destuser.sessions[packet.resource] == session then | |
226 | log("warn", "core", "Attempt to send stanza to self, dropping..."); | |
227 | else | |
228 | print("...sending..."); | |
229 | copas_send(destuser.sessions[packet.resource].conn, tostring(packet.data)); | |
230 | print("...sent") | |
231 | end | |
232 | elseif packet.data.name == "message" then | |
233 | print(" ...will be stored offline"); | |
234 | offlinemessage.new(packet.node, packet.host, packet.data); | |
235 | elseif packet.data.name == "iq" then | |
236 | print(" ...is an iq"); | |
237 | send(st.reply(packet.data) | |
238 | :tag("error", { type = "cancel" }) | |
239 | :tag("service-unavailable", { xmlns = "urn:ietf:params:xml:ns:xmpp-stanzas" })); | |
240 | end | |
241 | print(" ...removing from send buffer"); | |
242 | sendbuffer.external[n] = nil; | |
243 | end | |
244 | end | |
245 | end | |
246 | ||
247 | if #sendbuffer > 0 then | |
248 | for n, data in ipairs(sendbuffer) do | |
249 | print "Sending..." | |
250 | copas_send(conn, data); | |
251 | print "Sent" | |
252 | sendbuffer[n] = nil; | |
253 | end | |
254 | end | |
255 | print "Receiving..." | |
256 | repeat | |
257 | reqdata, sktmsg = copas_receive(conn, 1); | |
258 | if sktmsg == 'wantread' then | |
259 | print("Received... wantread"); | |
260 | --socket.select({conn}, nil) | |
261 | --print("Socket ready now..."); | |
262 | elseif sktmsg then | |
263 | print("Received socket message:", sktmsg); | |
264 | end | |
265 | until reqdata or sktmsg == "closed"; | |
266 | print("Received", tostring(reqdata)); | |
267 | end | |
268 | log("info", "core", "Client disconnected, connection closed"); | |
269 | end | |
270 | ||
271 | server = socket.bind("*", 5223) | |
272 | assert(server, "Failed to bind to socket") | |
273 | copas.addserver(server, handler) | |
274 | ||
275 | copas.loop(); |