prosody
b8787e859fd2
main.lua

Find changesets by keywords by author, files, the commit message, revision number or hash, or revset expression.

Software / code / prosody

Comparison

main.lua @ 1:b8787e859fd2

Switched to new connection framework, courtesy of the luadch project Now supports SSL on 5223 Beginning support for presence (aka. the proper routing of stanzas)
author matthew
date Sun, 24 Aug 2008 01:51:02 +0000
parent 0:3e3171b59028
child 2:9bb397205f26
comparison
equal deleted inserted replaced
0:3e3171b59028 1:b8787e859fd2
1 require "luarocks.require" 1 require "luarocks.require"
2 2
3 require "copas" 3 server = require "server"
4 require "socket" 4 require "socket"
5 require "ssl" 5 require "ssl"
6 require "lxp" 6 require "lxp"
7 7
8 function log(type, area, message) 8 function log(type, area, message)
9 print(type, area, message); 9 print(type, area, message);
10 end 10 end
11 11
12 require "core.stanza_dispatch" 12 require "core.stanza_dispatch"
13 init_xmlhandlers = require "core.xmlhandlers"
13 require "core.rostermanager" 14 require "core.rostermanager"
14 require "core.offlinemessage" 15 require "core.offlinemessage"
16 require "core.usermanager"
15 require "util.stanza" 17 require "util.stanza"
16 require "util.jid" 18 require "util.jid"
17 19
18 -- Locals for faster access -- 20 -- Locals for faster access --
19 local t_insert = table.insert; 21 local t_insert = table.insert;
22 local m_random = math.random; 24 local m_random = math.random;
23 local format = string.format; 25 local format = string.format;
24 local st = stanza; 26 local st = stanza;
25 ------------------------------ 27 ------------------------------
26 28
27 users = {}; 29 sessions = {};
28 hosts = { 30 hosts = {
29 ["localhost"] = { 31 ["localhost"] = {
30 type = "local"; 32 type = "local";
31 connected = true; 33 connected = true;
32 sessions = {}; 34 sessions = {};
38 }; 40 };
39 } 41 }
40 42
41 local hosts, users = hosts, users; 43 local hosts, users = hosts, users;
42 44
43 local ssl_ctx, msg = ssl.newcontext { mode = "server", protocol = "sslv23", key = "/home/matthew/ssl_cert/server.key", 45 --local ssl_ctx, msg = ssl.newcontext { mode = "server", protocol = "sslv23", key = "/home/matthew/ssl_cert/server.key",
46 -- certificate = "/home/matthew/ssl_cert/server.crt", capath = "/etc/ssl/certs", verify = "none", }
47 --
48 --if not ssl_ctx then error("Failed to initialise SSL/TLS support: "..tostring(msg)); end
49
50
51 local ssl_ctx = { 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", } 52 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 53
48 54
49 function connect_host(host) 55 function connect_host(host)
50 hosts[host] = { type = "remote", sendbuffer = {} }; 56 hosts[host] = { type = "remote", sendbuffer = {} };
51 end 57 end
52 58
53 function handler(conn) 59 local function send_to(session, to, stanza)
54 local copas_receive, copas_send = copas.receive, copas.send; 60 local node, host, resource = jid.split(to);
55 local reqdata, sktmsg; 61 if not hosts[host] then
56 local session = { sendbuffer = { external = {} }, conn = conn, notopen = true, priority = 0 } 62 -- s2s
63 elseif hosts[host].type == "local" then
64 print(" ...is to a local user")
65 local destuser = hosts[host].sessions[node];
66 if destuser and destuser.sessions then
67 if not destuser.sessions[resource] then
68 local best_session;
69 for resource, session in pairs(destuser.sessions) do
70 if not best_session then best_session = session;
71 elseif session.priority >= best_session.priority and session.priority >= 0 then
72 best_session = session;
73 end
74 end
75 if not best_session then
76 offlinemessage.new(node, host, stanza);
77 else
78 print("resource '"..resource.."' was not online, have chosen to send to '"..best_session.username.."@"..best_session.host.."/"..best_session.resource.."'");
79 resource = best_session.resource;
80 end
81 end
82 if destuser.sessions[resource] == session then
83 log("warn", "core", "Attempt to send stanza to self, dropping...");
84 else
85 print("...sending...", tostring(stanza));
86 --destuser.sessions[resource].conn.write(tostring(data));
87 print(" to conn ", destuser.sessions[resource].conn);
88 destuser.sessions[resource].conn.write(tostring(stanza));
89 print("...sent")
90 end
91 elseif stanza.name == "message" then
92 print(" ...will be stored offline");
93 offlinemessage.new(node, host, stanza);
94 elseif stanza.name == "iq" then
95 print(" ...is an iq");
96 session.send(st.reply(stanza)
97 :tag("error", { type = "cancel" })
98 :tag("service-unavailable", { xmlns = "urn:ietf:params:xml:ns:xmpp-stanzas" }));
99 end
100 print(" ...done routing");
101 end
102 end
103
104 function handler(conn, data, err)
105 local session = sessions[conn];
106
107 if not session then
108 sessions[conn] = { conn = conn, notopen = true, priority = 0 };
109 session = sessions[conn];
110
111 -- Logging functions --
112
113 local mainlog, log = log;
114 do
115 local conn_name = tostring(conn):match("%w+$");
116 log = function (type, area, message) mainlog(type, conn_name, message); end
117 end
118 local print = function (...) log("info", "core", t_concatall({...}, "\t")); end
119 session.log = log;
120
121 -- -- --
122
123 -- Send buffers --
124
125 local send = function (data) print("Sending...", tostring(data)); conn.write(tostring(data)); end;
126 session.send, session.send_to = send, send_to;
127
128 print("Client connected");
129
130 session.stanza_dispatch = init_stanza_dispatcher(session);
131 session.xml_handlers = init_xmlhandlers(session);
132 session.parser = lxp.new(session.xml_handlers, ":");
133
134 function session.disconnect(err)
135 print("Disconnected: "..err);
136 end
137 end
138 if data then
139 session.parser:parse(data);
140 end
141
142 --log("info", "core", "Client disconnected, connection closed");
143 end
144
145 function disconnect(conn, err)
146 sessions[conn].disconnect(err);
147 end
148
149 print("ssl_ctx:", type(ssl_ctx));
150
151 setmetatable(_G, { __index = function (t, k) print("WARNING: ATTEMPT TO READ A NIL GLOBAL!!!", k); error("Attempt to read a non-existent global. Naughty boy.", 2); end, __newindex = function (t, k, v) print("ATTEMPT TO SET A GLOBAL!!!!", tostring(k).." = "..tostring(v)); error("Attempt to set a global. Naughty boy.", 2); end }) --]][][[]][];
57 152
58 153
59 -- Logging functions -- 154 local protected_handler = function (...) local success, ret = pcall(handler, ...); if not success then print("ERROR on "..tostring((select(1, ...)))..": "..ret); end end;
60 155
61 local mainlog, log = log; 156 print( server.add( { listener = protected_handler, disconnect = disconnect }, 5222, "*", 1, nil ) ) -- server.add will send a status message
62 do 157 print( server.add( { listener = protected_handler, disconnect = disconnect }, 5223, "*", 1, ssl_ctx ) ) -- server.add will send a status message
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 158
69 -- -- -- 159 server.loop();
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();