Mercurial > prosody-hg

comparison core/s2smanager.lua @ 282:80e7de32b618

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Merging my new SASL code with Waqas' adjusted saslauth module.
author Tobias Markmann <tm@ayena.de>
date Sat, 15 Nov 2008 13:47:17 +0100
parents 3cfac0e5e6ca
children 9439362caacc
comparison
equal deleted inserted replaced
280:516f4c901991 282:80e7de32b618
2 local hosts = hosts; 2 local hosts = hosts;
3 local sessions = sessions; 3 local sessions = sessions;
4 local socket = require "socket"; 4 local socket = require "socket";
5 local format = string.format; 5 local format = string.format;
6 local t_insert = table.insert; 6 local t_insert = table.insert;
7 local get_traceback = debug.traceback;
7 local tostring, pairs, ipairs, getmetatable, print, newproxy, error, tonumber 8 local tostring, pairs, ipairs, getmetatable, print, newproxy, error, tonumber
8 = tostring, pairs, ipairs, getmetatable, print, newproxy, error, tonumber; 9 = tostring, pairs, ipairs, getmetatable, print, newproxy, error, tonumber;
9 10
10 local connlisteners_get = require "net.connlisteners".get; 11 local connlisteners_get = require "net.connlisteners".get;
11 local wraptlsclient = require "net.server".wraptlsclient; 12 local wraptlsclient = require "net.server".wraptlsclient;
12 local modulemanager = require "core.modulemanager"; 13 local modulemanager = require "core.modulemanager";
14 local st = require "stanza";
15 local stanza = st.stanza;
13 16
14 local uuid_gen = require "util.uuid".generate; 17 local uuid_gen = require "util.uuid".generate;
15 18
16 local logger_init = require "util.logger".init; 19 local logger_init = require "util.logger".init;
17 20
19 22
20 local md5_hash = require "util.hashes".md5; 23 local md5_hash = require "util.hashes".md5;
21 24
22 local dialback_secret = "This is very secret!!! Ha!"; 25 local dialback_secret = "This is very secret!!! Ha!";
23 26
24 local srvmap = { ["gmail.com"] = "talk.google.com", ["identi.ca"] = "longlance.controlezvous.ca", ["cdr.se"] = "jabber.cdr.se" }; 27 local srvmap = { ["gmail.com"] = "talk.google.com", ["identi.ca"] = "hampton.controlezvous.ca", ["cdr.se"] = "jabber.cdr.se" };
25 28
26 module "s2smanager" 29 module "s2smanager"
27 30
28 function connect_host(from_host, to_host)
29 end
30
31 function send_to_host(from_host, to_host, data) 31 function send_to_host(from_host, to_host, data)
32 local host = hosts[to_host]; 32 local host = hosts[from_host].s2sout[to_host];
33 if host then 33 if host then
34 -- Write to connection 34 -- We have a connection to this host already
35 if host.type == "s2sout_unauthed" and not host.notopen and not host.dialback_key then 35 if host.type == "s2sout_unauthed" then
36 log("debug", "trying to send over unauthed s2sout to "..to_host..", authing it now..."); 36 host.log("debug", "trying to send over unauthed s2sout to "..to_host..", authing it now...");
37 initiate_dialback(host); 37 if not host.notopen and not host.dialback_key then
38 if not host.sendq then host.sendq = { data }; 38 host.log("debug", "dialback had not been initiated");
39 else t_insert(host.sendq, data); end 39 initiate_dialback(host);
40 end
41
42 -- Queue stanza until we are able to send it
43 if host.sendq then t_insert(host.sendq, data);
44 else host.sendq = { data }; end
45 elseif host.type == "local" or host.type == "component" then
46 log("error", "Trying to send a stanza to ourselves??")
47 log("error", "Traceback: %s", get_traceback());
48 log("error", "Stanza: %s", tostring(data));
40 else 49 else
41 log("debug", "going to send stanza to "..to_host.." from "..from_host); 50 (host.log or log)("debug", "going to send stanza to "..to_host.." from "..from_host);
42 -- FIXME 51 -- FIXME
43 if hosts[to_host].from_host ~= from_host then log("error", "WARNING! This might, possibly, be a bug, but it might not..."); end 52 if host.from_host ~= from_host then
44 hosts[to_host].sends2s(data); 53 log("error", "WARNING! This might, possibly, be a bug, but it might not...");
45 log("debug", "stanza sent over "..hosts[to_host].type); 54 log("error", "We are going to send from %s instead of %s", host.from_host, from_host);
55 end
56 host.sends2s(data);
57 host.log("debug", "stanza sent over "..host.type);
46 end 58 end
47 else 59 else
48 log("debug", "opening a new outgoing connection for this stanza"); 60 log("debug", "opening a new outgoing connection for this stanza");
49 local host_session = new_outgoing(from_host, to_host); 61 local host_session = new_outgoing(from_host, to_host);
50 -- Store in buffer 62 -- Store in buffer
51 host_session.sendq = { data }; 63 host_session.sendq = { data };
52 end 64 end
53 end 65 end
54 66
55 function disconnect_host(host)
56
57 end
58
59 local open_sessions = 0; 67 local open_sessions = 0;
60 68
61 function new_incoming(conn) 69 function new_incoming(conn)
62 local session = { conn = conn, type = "s2sin_unauthed", direction = "incoming" }; 70 local session = { conn = conn, type = "s2sin_unauthed", direction = "incoming" };
63 if true then 71 if true then
70 return session; 78 return session;
71 end 79 end
72 80
73 function new_outgoing(from_host, to_host) 81 function new_outgoing(from_host, to_host)
74 local host_session = { to_host = to_host, from_host = from_host, notopen = true, type = "s2sout_unauthed", direction = "outgoing" }; 82 local host_session = { to_host = to_host, from_host = from_host, notopen = true, type = "s2sout_unauthed", direction = "outgoing" };
75 hosts[to_host] = host_session; 83 hosts[from_host].s2sout[to_host] = host_session;
76 local cl = connlisteners_get("xmppserver"); 84 local cl = connlisteners_get("xmppserver");
77 85
78 local conn, handler = socket.tcp() 86 local conn, handler = socket.tcp()
79 87
88 --FIXME: Below parameters (ports/ip) are incorrect (use SRV)
89 to_host = srvmap[to_host] or to_host;
90
91 conn:settimeout(0);
92 local success, err = conn:connect(to_host, 5269);
93 if not success then
94 log("warn", "s2s connect() failed: %s", err);
95 end
96
97 conn = wraptlsclient(cl, conn, to_host, 5269, 0, 1, hosts[from_host].ssl_ctx );
98 host_session.conn = conn;
80 99
81 -- Register this outgoing connection so that xmppserver_listener knows about it 100 -- Register this outgoing connection so that xmppserver_listener knows about it
82 -- otherwise it will assume it is a new incoming connection 101 -- otherwise it will assume it is a new incoming connection
83 cl.register_outgoing(conn, host_session); 102 cl.register_outgoing(conn, host_session);
84 103
85 --FIXME: Below parameters (ports/ip) are incorrect (use SRV)
86 to_host = srvmap[to_host] or to_host;
87 conn:settimeout(0.1);
88 conn:connect(to_host, 5269);
89 conn = wraptlsclient(cl, conn, to_host, 5269, 0, 1, hosts[from_host].ssl_ctx );
90 host_session.conn = conn;
91
92 do 104 do
93 local conn_name = "s2sout"..tostring(conn):match("[a-f0-9]*$"); 105 local conn_name = "s2sout"..tostring(conn):match("[a-f0-9]*$");
94 host_session.log = logger_init(conn_name); 106 host_session.log = logger_init(conn_name);
95 end 107 end
96 108
101 113
102 return host_session; 114 return host_session;
103 end 115 end
104 116
105 function streamopened(session, attr) 117 function streamopened(session, attr)
106 session.log("debug", "s2s stream opened");
107 local send = session.sends2s; 118 local send = session.sends2s;
108 119
109 session.version = tonumber(attr.version) or 0; 120 session.version = tonumber(attr.version) or 0;
110 if session.version >= 1.0 and not (attr.to and attr.from) then 121 if session.version >= 1.0 and not (attr.to and attr.from) then
111 print("to: "..tostring(attr.to).." from: "..tostring(attr.from)); 122 print("to: "..tostring(attr.to).." from: "..tostring(attr.from));
122 session.from_host = attr.from; 133 session.from_host = attr.from;
123 134
124 session.streamid = uuid_gen(); 135 session.streamid = uuid_gen();
125 print(session, session.from_host, "incoming s2s stream opened"); 136 print(session, session.from_host, "incoming s2s stream opened");
126 send("<?xml version='1.0'?>"); 137 send("<?xml version='1.0'?>");
127 send(format("<stream:stream xmlns='jabber:server' xmlns:db='jabber:server:dialback' xmlns:stream='http://etherx.jabber.org/streams' id='%s' from='%s'>", session.streamid, session.to_host)); 138 send(stanza("stream:stream", { xmlns='jabber:server', ["xmlns:db"]='jabber:server:dialback', ["xmlns:stream"]='http://etherx.jabber.org/streams', id=session.streamid, from=session.to_host }):top_tag());
128 elseif session.direction == "outgoing" then 139 elseif session.direction == "outgoing" then
129 -- If we are just using the connection for verifying dialback keys, we won't try and auth it 140 -- If we are just using the connection for verifying dialback keys, we won't try and auth it
130 if not attr.id then error("stream response did not give us a streamid!!!"); end 141 if not attr.id then error("stream response did not give us a streamid!!!"); end
131 session.streamid = attr.id; 142 session.streamid = attr.id;
132 143
145 for _, feature in ipairs(features) do 156 for _, feature in ipairs(features) do
146 send(tostring(feature)); 157 send(tostring(feature));
147 end 158 end
148 159
149 send("</stream:features>");]] 160 send("</stream:features>");]]
150 log("info", "s2s stream opened successfully"); 161
151 session.notopen = nil; 162 session.notopen = nil;
152 end 163 end
153 164
154 function initiate_dialback(session) 165 function initiate_dialback(session)
155 -- generate dialback key 166 -- generate dialback key
192 function session.send(data) send_to_host(to, from, data); end 203 function session.send(data) send_to_host(to, from, data); end
193 204
194 205
195 if session.direction == "outgoing" then 206 if session.direction == "outgoing" then
196 if sendq then 207 if sendq then
197 session.log("debug", "sending queued stanzas across new outgoing connection to "..session.to_host); 208 session.log("debug", "sending "..#sendq.." queued stanzas across new outgoing connection to "..session.to_host);
198 for i, data in ipairs(sendq) do 209 for i, data in ipairs(sendq) do
199 send(data); 210 send(data);
200 sendq[i] = nil; 211 sendq[i] = nil;
201 end 212 end
202 session.sendq = nil; 213 session.sendq = nil;
205 end 216 end
206 217
207 function destroy_session(session) 218 function destroy_session(session)
208 (session.log or log)("info", "Destroying "..tostring(session.direction).." session "..tostring(session.from_host).."->"..tostring(session.to_host)); 219 (session.log or log)("info", "Destroying "..tostring(session.direction).." session "..tostring(session.from_host).."->"..tostring(session.to_host));
209 if session.direction == "outgoing" then 220 if session.direction == "outgoing" then
210 hosts[session.to_host] = nil; 221 hosts[session.from_host].s2sout[session.to_host] = nil;
211 end 222 end
212 session.conn = nil; 223 session.conn = nil;
213 session.disconnect = nil; 224 session.disconnect = nil;
214 for k in pairs(session) do 225 for k in pairs(session) do
215 if k ~= "trace" then 226 if k ~= "trace" then