diff options
Diffstat (limited to 'xmpp-vala/src/module')
| -rw-r--r-- | xmpp-vala/src/module/message/stanza.vala | 4 | ||||
| -rw-r--r-- | xmpp-vala/src/module/stanza_error.vala | 7 | ||||
| -rw-r--r-- | xmpp-vala/src/module/xep/0047_in_band_bytestreams.vala | 468 | ||||
| -rw-r--r-- | xmpp-vala/src/module/xep/0048_bookmarks/conference.vala | 4 | ||||
| -rw-r--r-- | xmpp-vala/src/module/xep/0166_jingle.vala | 494 | ||||
| -rw-r--r-- | xmpp-vala/src/module/xep/0234_jingle_file_transfer.vala | 130 | ||||
| -rw-r--r-- | xmpp-vala/src/module/xep/0261_jingle_in_band_bytestreams.vala | 77 |
7 files changed, 1178 insertions, 6 deletions
diff --git a/xmpp-vala/src/module/message/stanza.vala b/xmpp-vala/src/module/message/stanza.vala index 640f2796..053c44dd 100644 --- a/xmpp-vala/src/module/message/stanza.vala +++ b/xmpp-vala/src/module/message/stanza.vala @@ -18,7 +18,7 @@ public class MessageStanza : Xmpp.Stanza { public string body { get { StanzaNode? body_node = stanza.get_subnode(NODE_BODY); - return body_node == null? null : body_node.get_string_content(); + return body_node == null ? null : body_node.get_string_content(); } set { StanzaNode? body_node = stanza.get_subnode(NODE_BODY); @@ -65,4 +65,4 @@ public abstract class MessageFlag : Object { public abstract string get_id(); } -}
\ No newline at end of file +} diff --git a/xmpp-vala/src/module/stanza_error.vala b/xmpp-vala/src/module/stanza_error.vala index 651e8d2b..c108b02a 100644 --- a/xmpp-vala/src/module/stanza_error.vala +++ b/xmpp-vala/src/module/stanza_error.vala @@ -82,8 +82,8 @@ namespace Xmpp { public ErrorStanza.bad_request(string? human_readable = null) { this.build(TYPE_MODIFY, CONDITION_BAD_REQUEST, human_readable, null); } - public ErrorStanza.feature_not_implemented(StanzaNode? application_condition = null) { - this.build(TYPE_MODIFY, CONDITION_FEATURE_NOT_IMPLEMENTED, null, application_condition); + public ErrorStanza.feature_not_implemented(string? human_readable = null) { + this.build(TYPE_MODIFY, CONDITION_FEATURE_NOT_IMPLEMENTED, human_readable, null); } public ErrorStanza.item_not_found(StanzaNode? application_condition = null) { this.build(TYPE_CANCEL, CONDITION_ITEM_NOT_FOUND, null, application_condition); @@ -91,6 +91,9 @@ namespace Xmpp { public ErrorStanza.not_acceptable(string? human_readable = null) { this.build(TYPE_MODIFY, CONDITION_NOT_ACCEPTABLE, human_readable, null); } + public ErrorStanza.not_allowed(string? human_readable = null) { + this.build(TYPE_CANCEL, CONDITION_NOT_ALLOWED, human_readable, null); + } public ErrorStanza.service_unavailable() { this.build(TYPE_CANCEL, CONDITION_SERVICE_UNAVAILABLE, null, null); } diff --git a/xmpp-vala/src/module/xep/0047_in_band_bytestreams.vala b/xmpp-vala/src/module/xep/0047_in_band_bytestreams.vala new file mode 100644 index 00000000..89247780 --- /dev/null +++ b/xmpp-vala/src/module/xep/0047_in_band_bytestreams.vala @@ -0,0 +1,468 @@ +using Gee; +using Xmpp; +using Xmpp.Xep; + +namespace Xmpp.Xep.InBandBytestreams { + +private const string NS_URI = "http://jabber.org/protocol/ibb"; +private const int SEQ_MODULUS = 65536; + +public class Module : XmppStreamModule, Iq.Handler { + public static Xmpp.ModuleIdentity<Module> IDENTITY = new Xmpp.ModuleIdentity<Module>(NS_URI, "0047_in_band_bytestreams"); + + public override void attach(XmppStream stream) { + stream.add_flag(new Flag()); + stream.get_module(Iq.Module.IDENTITY).register_for_namespace(NS_URI, this); + } + public override void detach(XmppStream stream) { } + + public void on_iq_set(XmppStream stream, Iq.Stanza iq) { + // the iq module ensures that there's only one child node + StanzaNode? node = null; + node = (node != null) ? node : iq.stanza.get_subnode("open", NS_URI); + node = (node != null) ? node : iq.stanza.get_subnode("data", NS_URI); + node = (node != null) ? node : iq.stanza.get_subnode("close", NS_URI); + if (node == null) { + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("unknown IBB action"))); + return; + } + string? sid = node.get_attribute("sid"); + if (sid == null) { + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("missing sid"))); + return; + } + Connection? conn = stream.get_flag(Flag.IDENTITY).get_connection(sid); + if (node.name == "open") { + if (conn == null) { + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.not_acceptable("unexpected IBB connection"))); + return; + } + if (conn.state != WAITING_FOR_CONNECT) { + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("IBB open for already open connection"))); + return; + } + conn.handle_open(stream, node, iq); + } else { + if (conn == null || conn.state != Connection.State.CONNECTED) { + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.item_not_found())); + return; + } + if (node.name == "close") { + conn.handle_close(stream, node, iq); + } else { + conn.handle_data(stream, node, iq); + } + } + } + + public override string get_ns() { return NS_URI; } + public override string get_id() { return IDENTITY.id; } +} + +public class Connection : IOStream { + // TODO(hrxi): Fix reference cycle + public class Input : InputStream { + private Connection connection; + public Input(Connection connection) { + this.connection = connection; + } + public override ssize_t read(uint8[] buffer, Cancellable? cancellable = null) throws IOError { + throw new IOError.NOT_SUPPORTED("can't do non-async reads on in-band bytestreams"); + } + public override async ssize_t read_async(uint8[]? buffer, int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError { + return yield connection.read_async(buffer, io_priority, cancellable); + } + public override bool close(Cancellable? cancellable = null) throws IOError { + return connection.close_read(cancellable); + } + public override async bool close_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError { + return yield connection.close_read_async(io_priority, cancellable); + } + } + public class Output : OutputStream { + private Connection connection; + public Output(Connection connection) { + this.connection = connection; + } + public override ssize_t write(uint8[] buffer, Cancellable? cancellable = null) throws IOError { + throw new IOError.NOT_SUPPORTED("can't do non-async writes on in-band bytestreams"); + } + public override async ssize_t write_async(uint8[]? buffer, int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError { + return yield connection.write_async(buffer, io_priority, cancellable); + } + public override bool close(Cancellable? cancellable = null) throws IOError { + return connection.close_write(cancellable); + } + public override async bool close_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError { + return yield connection.close_write_async(io_priority, cancellable); + } + } + + private Input input; + private Output output; + public override InputStream input_stream { get { return input; } } + public override OutputStream output_stream { get { return output; } } + + public enum State { + WAITING_FOR_CONNECT, + CONNECTING, + CONNECTED, + DISCONNECTING, + DISCONNECTED, + ERROR, + } + public State state { get; private set; } + Jid receiver_full_jid; + public string sid { get; private set; } + int block_size; + int local_seq = 0; + int remote_ack = 0; + internal int remote_seq = 0; + + bool input_closed = false; + bool output_closed = false; + + // ERROR + string? error = null; + + XmppStream stream; + + int read_callback_priority; + Cancellable? read_callback_cancellable = null; + ulong read_callback_cancellable_id; + SourceFunc? read_callback = null; + int write_callback_priority; + SourceFunc? write_callback = null; + ulong write_callback_cancellable_id; + Cancellable? write_callback_cancellable = null; + // Need `Bytes` instead of `uint8[]` because the latter doesn't work in + // parameter position of `LinkedList`. + LinkedList<Bytes> received = new LinkedList<Bytes>(); + + private Connection(XmppStream stream, Jid receiver_full_jid, string sid, int block_size, bool initiate) { + this.stream = stream; + this.receiver_full_jid = receiver_full_jid; + this.sid = sid; + this.block_size = block_size; + this.state = initiate ? State.CONNECTING : State.WAITING_FOR_CONNECT; + + input = new Input(this); + output = new Output(this); + } + + public void set_read_callback(SourceFunc callback, Cancellable? cancellable, int io_priority) throws IOError { + if (read_callback != null) { + throw new IOError.PENDING("only one async read is permitted at a time on an in-band bytestream"); + } + if (cancellable != null) { + read_callback_cancellable_id = cancellable.connect(trigger_read_callback); + } + read_callback = callback; + read_callback_cancellable = cancellable; + read_callback_priority = io_priority; + } + public void set_write_callback(SourceFunc callback, Cancellable? cancellable, int io_priority) throws IOError { + if (write_callback != null) { + throw new IOError.PENDING("only one async write is permitted at a time on an in-band bytestream"); + } + if (cancellable != null) { + write_callback_cancellable_id = cancellable.connect(trigger_write_callback); + } + write_callback = callback; + write_callback_cancellable = cancellable; + write_callback_priority = io_priority; + } + public void trigger_read_callback() { + if (read_callback != null) { + Idle.add((owned) read_callback, read_callback_priority); + read_callback = null; + if (read_callback_cancellable != null) { + read_callback_cancellable.disconnect(read_callback_cancellable_id); + } + read_callback_cancellable = null; + } + } + public void trigger_write_callback() { + if (write_callback != null) { + Idle.add((owned) write_callback, write_callback_priority); + write_callback = null; + if (write_callback_cancellable != null) { + write_callback_cancellable.disconnect(write_callback_cancellable_id); + } + write_callback_cancellable = null; + } + } + + public async ssize_t read_async(uint8[]? buffer, int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError { + while (true) { + if (cancellable != null) { + cancellable.set_error_if_cancelled(); + } + if (input_closed) { + return 0; + } + Bytes? chunk = received.poll(); + if (chunk != null) { + int read = int.min(buffer.length, chunk.length); + for (int i = 0; i < read; i++) { + buffer[i] = chunk[i]; + } + if (buffer.length < chunk.length) { + received.offer_head(chunk[buffer.length:chunk.length]); + } + return read; + } + if (state == DISCONNECTED) { + return 0; + } + set_read_callback(read_async.callback, cancellable, io_priority); + yield; + } + } + + public async ssize_t write_async(uint8[]? buffer, int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError { + while (state == WAITING_FOR_CONNECT || state == CONNECTING) { + if (cancellable != null) { + cancellable.set_error_if_cancelled(); + } + set_write_callback(write_async.callback, cancellable, io_priority); + yield; + } + throw_if_closed(); + assert(state == CONNECTED); + // TODO(hrxi): merging? + int seq = local_seq; + local_seq = (local_seq + 1) % SEQ_MODULUS; + if (buffer.length > block_size) { + buffer = buffer[0:block_size]; + } + StanzaNode data = new StanzaNode.build("data", NS_URI) + .add_self_xmlns() + .put_attribute("sid", sid) + .put_attribute("seq", seq.to_string()) + .put_node(new StanzaNode.text(Base64.encode(buffer))); + Iq.Stanza iq = new Iq.Stanza.set(data) { to=receiver_full_jid }; + set_write_callback(write_async.callback, cancellable, io_priority); + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => { + if (iq.is_error()) { + set_error("sending failed"); + } else if (remote_ack != seq) { + set_error("out of order acks"); + } else { + remote_ack = (remote_ack + 1) % SEQ_MODULUS; + if (local_seq == remote_ack) { + trigger_write_callback(); + } + } + }); + yield; + if (cancellable != null) { + cancellable.set_error_if_cancelled(); + } + throw_if_error(); + return buffer.length; + } + + public bool close_read(Cancellable? cancellable = null) { + input_closed = true; + if (!output_closed) { + return true; + } + return close_impl(cancellable); + } + public async bool close_read_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError { + input_closed = true; + if (!output_closed) { + return true; + } + return yield close_async_impl(io_priority, cancellable); + } + public bool close_write(Cancellable? cancellable = null) { + output_closed = true; + if (!input_closed) { + return true; + } + return close_impl(cancellable); + } + public async bool close_write_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError { + output_closed = true; + if (!input_closed) { + return true; + } + return yield close_async_impl(io_priority, cancellable); + } + delegate void OnClose(bool success); + private bool close_impl(Cancellable? cancellable = null, OnClose? on_close = null) { + if (state == DISCONNECTING || state == DISCONNECTED || state == ERROR) { + on_close(true); + return true; + } + if (state == WAITING_FOR_CONNECT) { + state = DISCONNECTED; + stream.get_flag(Flag.IDENTITY).remove_connection(this); + trigger_read_callback(); + on_close(true); + return true; + } + state = DISCONNECTING; + StanzaNode close = new StanzaNode.build("close", NS_URI) + .add_self_xmlns() + .put_attribute("sid", sid); + Iq.Stanza iq = new Iq.Stanza.set(close) { to=receiver_full_jid }; + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => { + assert(state == DISCONNECTING); + if (iq.is_error()) { + set_error("disconnecting failed"); + } else { + state = DISCONNECTED; + } + stream.get_flag(Flag.IDENTITY).remove_connection(this); + trigger_read_callback(); + on_close(!iq.is_error()); + }); + return true; + } + private async bool close_async_impl(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError { + SourceFunc callback = close_async_impl.callback; + close_impl(cancellable, () => { Idle.add((owned) callback); }); + yield; + throw_if_error(); + return true; + } + + public static Connection create(XmppStream stream, Jid receiver_full_jid, string sid, int block_size, bool initiate) { + Connection conn = new Connection(stream, receiver_full_jid, sid, block_size, initiate); + if (initiate) { + StanzaNode open = new StanzaNode.build("open", NS_URI) + .add_self_xmlns() + .put_attribute("block-size", block_size.to_string()) + .put_attribute("sid", sid); + + Iq.Stanza iq = new Iq.Stanza.set(open) { to=receiver_full_jid }; + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => { + if (conn.state != CONNECTING) { + assert(conn.state != CONNECTED); + return; + } + if (!iq.is_error()) { + conn.state = CONNECTED; + stream.get_flag(Flag.IDENTITY).add_connection(conn); + conn.trigger_write_callback(); + } else { + conn.set_error("connection failed"); + } + }); + } else { + stream.get_flag(Flag.IDENTITY).add_connection(conn); + } + return conn; + } + + void throw_if_error() throws IOError { + if (state == ERROR) { + throw new IOError.FAILED(error); + } + } + + void throw_if_closed() throws IOError { + throw_if_error(); + if (state == DISCONNECTING || state == DISCONNECTED) { + throw new IOError.CLOSED("can't read/write on a closed connection"); + } + } + + void set_error(string error) { + if (state != WAITING_FOR_CONNECT && state != DISCONNECTING && state != DISCONNECTED && state != ERROR) { + close_async.begin(); + } + state = ERROR; + this.error = error; + stream.get_flag(Flag.IDENTITY).remove_connection(this); + } + + public void handle_open(XmppStream stream, StanzaNode open, Iq.Stanza iq) { + assert(state == WAITING_FOR_CONNECT); + int block_size = open.get_attribute_int("block-size"); + string? stanza = open.get_attribute("stanza"); + if (block_size < 0 || (stanza != null && stanza != "iq" && stanza != "message")) { + set_error("invalid open"); + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("missing block_size or invalid stanza"))); + return; + } + if (stanza != null && stanza != "iq") { + set_error("invalid open"); + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.feature_not_implemented("cannot use message stanzas for IBB"))); + return; + } + if (block_size > this.block_size) { + set_error("invalid open"); + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.build(ErrorStanza.TYPE_CANCEL, ErrorStanza.CONDITION_RESOURCE_CONSTRAINT, "opening a connection with a greater than negotiated/acceptable block size", null))); + return; + } + this.block_size = block_size; + state = CONNECTED; + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq)); + trigger_write_callback(); + } + public void handle_data(XmppStream stream, StanzaNode data, Iq.Stanza iq) { + assert(state == CONNECTED); + if (input_closed) { + set_error("unexpected data"); + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.not_allowed("unexpected data"))); + return; + } + int seq = data.get_attribute_int("seq"); + // TODO(hrxi): return an error on malformed base64 (need to do this + // according to the xep) + uint8[] content = Base64.decode(data.get_string_content()); + if (content.length > block_size) { + set_error("data longer than negotiated block size"); + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.bad_request("data longer than negotiated block size"))); + return; + } + if (seq < 0 || seq != remote_seq) { + set_error("out of order data packets"); + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.build(ErrorStanza.TYPE_CANCEL, ErrorStanza.CONDITION_UNEXPECTED_REQUEST, "out of order data packets", null))); + return; + } + remote_seq = (remote_seq + 1) % SEQ_MODULUS; + + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq)); + if (content.length != 0) { + received.offer(new Bytes.take(content)); + trigger_read_callback(); + } + } + public void handle_close(XmppStream stream, StanzaNode close, Iq.Stanza iq) { + assert(state == CONNECTED); + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq)); + stream.get_flag(Flag.IDENTITY).remove_connection(this); + input_closed = true; + output_closed = true; + state = DISCONNECTED; + + trigger_read_callback(); + } +} + + +public class Flag : XmppStreamFlag { + public static FlagIdentity<Flag> IDENTITY = new FlagIdentity<Flag>(NS_URI, "in_band_bytestreams"); + + private HashMap<string, Connection> active = new HashMap<string, Connection>(); + + public void add_connection(Connection conn) { + active[conn.sid] = conn; + } + public Connection? get_connection(string sid) { + return active.has_key(sid) ? active[sid] : null; + } + public void remove_connection(Connection conn) { + active.unset(conn.sid); + } + + public override string get_ns() { return NS_URI; } + public override string get_id() { return IDENTITY.id; } +} + +} diff --git a/xmpp-vala/src/module/xep/0048_bookmarks/conference.vala b/xmpp-vala/src/module/xep/0048_bookmarks/conference.vala index c00d8f86..7f80490b 100644 --- a/xmpp-vala/src/module/xep/0048_bookmarks/conference.vala +++ b/xmpp-vala/src/module/xep/0048_bookmarks/conference.vala @@ -36,7 +36,7 @@ public class Conference : Object { public string? nick { get { StanzaNode? nick_node = stanza_node.get_subnode(NODE_NICK); - return nick_node == null? null : nick_node.get_string_content(); + return nick_node == null ? null : nick_node.get_string_content(); } set { StanzaNode? nick_node = stanza_node.get_subnode(NODE_NICK); @@ -56,7 +56,7 @@ public class Conference : Object { public string? password { get { StanzaNode? password_node = stanza_node.get_subnode(NODE_PASSWORD); - return password_node == null? null : password_node.get_string_content(); + return password_node == null ? null : password_node.get_string_content(); } set { StanzaNode? password_node = stanza_node.get_subnode(NODE_PASSWORD); diff --git a/xmpp-vala/src/module/xep/0166_jingle.vala b/xmpp-vala/src/module/xep/0166_jingle.vala new file mode 100644 index 00000000..7413ff4f --- /dev/null +++ b/xmpp-vala/src/module/xep/0166_jingle.vala @@ -0,0 +1,494 @@ +using Gee; +using Xmpp.Xep; +using Xmpp; + +namespace Xmpp.Xep.Jingle { + +private const string NS_URI = "urn:xmpp:jingle:1"; +private const string ERROR_NS_URI = "urn:xmpp:jingle:errors:1"; + +public errordomain IqError { + BAD_REQUEST, + NOT_ACCEPTABLE, + NOT_IMPLEMENTED, + OUT_OF_ORDER, +} + +void send_iq_error(IqError iq_error, XmppStream stream, Iq.Stanza iq) { + ErrorStanza error; + if (iq_error is IqError.BAD_REQUEST) { + error = new ErrorStanza.bad_request(iq_error.message); + } else if (iq_error is IqError.NOT_ACCEPTABLE) { + error = new ErrorStanza.not_acceptable(iq_error.message); + } else if (iq_error is IqError.NOT_IMPLEMENTED) { + error = new ErrorStanza.feature_not_implemented(iq_error.message); + } else if (iq_error is IqError.OUT_OF_ORDER) { + StanzaNode out_of_order = new StanzaNode.build("out-of-order", ERROR_NS_URI).add_self_xmlns(); + error = new ErrorStanza.build(ErrorStanza.TYPE_MODIFY, ErrorStanza.CONDITION_UNEXPECTED_REQUEST, iq_error.message, out_of_order); + } else { + assert_not_reached(); + } + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, error)); +} + +public errordomain Error { + GENERAL, + BAD_REQUEST, + INVALID_PARAMETERS, + UNSUPPORTED_TRANSPORT, + NO_SHARED_PROTOCOLS, + TRANSPORT_ERROR, +} + +StanzaNode get_single_node_anyns(StanzaNode parent, string node_name) throws IqError { + StanzaNode? result = null; + foreach (StanzaNode child in parent.get_all_subnodes()) { + if (child.name == node_name) { + if (result != null) { + throw new IqError.BAD_REQUEST(@"multiple $(node_name) nodes"); + } + result = child; + } + } + if (result == null) { + throw new IqError.BAD_REQUEST(@"missing $(node_name) node"); + } + return result; +} + +public class Module : XmppStreamModule, Iq.Handler { + public static Xmpp.ModuleIdentity<Module> IDENTITY = new Xmpp.ModuleIdentity<Module>(NS_URI, "0166_jingle"); + + private HashMap<string, ContentType> content_types = new HashMap<string, ContentType>(); + private HashMap<string, Transport> transports = new HashMap<string, Transport>(); + + public override void attach(XmppStream stream) { + stream.add_flag(new Flag()); + stream.get_module(ServiceDiscovery.Module.IDENTITY).add_feature(stream, NS_URI); + stream.get_module(Iq.Module.IDENTITY).register_for_namespace(NS_URI, this); + } + public override void detach(XmppStream stream) { } + + public void register_content_type(ContentType content_type) { + content_types[content_type.content_type_ns_uri()] = content_type; + } + public ContentType? get_content_type(string ns_uri) { + if (!content_types.has_key(ns_uri)) { + return null; + } + return content_types[ns_uri]; + } + public void register_transport(Transport transport) { + transports[transport.transport_ns_uri()] = transport; + } + public Transport? get_transport(string ns_uri) { + if (!transports.has_key(ns_uri)) { + return null; + } + return transports[ns_uri]; + } + public Transport? select_transport(XmppStream stream, TransportType type, Jid receiver_full_jid) { + foreach (Transport transport in transports.values) { + if (transport.transport_type() != type) { + continue; + } + // TODO(hrxi): prioritization + if (transport.is_transport_available(stream, receiver_full_jid)) { + return transport; + } + } + return null; + } + + private bool is_jingle_available(XmppStream stream, Jid full_jid) { + bool? has_jingle = stream.get_flag(ServiceDiscovery.Flag.IDENTITY).has_entity_feature(full_jid, NS_URI); + return has_jingle != null && has_jingle; + } + + public bool is_available(XmppStream stream, TransportType type, Jid full_jid) { + return is_jingle_available(stream, full_jid) && select_transport(stream, type, full_jid) != null; + } + + public Session create_session(XmppStream stream, TransportType type, Jid receiver_full_jid, Senders senders, string content_name, StanzaNode description) throws Error { + if (!is_jingle_available(stream, receiver_full_jid)) { + throw new Error.NO_SHARED_PROTOCOLS("No Jingle support"); + } + Transport? transport = select_transport(stream, type, receiver_full_jid); + if (transport == null) { + throw new Error.NO_SHARED_PROTOCOLS("No suitable transports"); + } + Jid? my_jid = stream.get_flag(Bind.Flag.IDENTITY).my_jid; + if (my_jid == null) { + throw new Error.GENERAL("Couldn't determine own JID"); + } + TransportParameters transport_params = transport.create_transport_parameters(); + Session session = new Session.initiate_sent(random_uuid(), type, transport_params, receiver_full_jid, content_name); + StanzaNode content = new StanzaNode.build("content", NS_URI) + .put_attribute("creator", "initiator") + .put_attribute("name", content_name) + .put_attribute("senders", senders.to_string()) + .put_node(description) + .put_node(transport_params.to_transport_stanza_node()); + StanzaNode jingle = new StanzaNode.build("jingle", NS_URI) + .add_self_xmlns() + .put_attribute("action", "session-initiate") + .put_attribute("initiator", my_jid.to_string()) + .put_attribute("sid", session.sid) + .put_node(content); + Iq.Stanza iq = new Iq.Stanza.set(jingle) { to=receiver_full_jid }; + + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq, (stream, iq) => { + // TODO(hrxi): handle errors + stream.get_flag(Flag.IDENTITY).add_session(session); + }); + + return session; + } + + public void handle_session_initiate(XmppStream stream, string sid, StanzaNode jingle, Iq.Stanza iq) throws IqError { + Gee.List<StanzaNode> contents = jingle.get_subnodes("content"); + if (contents.size == 0) { + throw new IqError.BAD_REQUEST("missing content node"); + } + if (contents.size > 1) { + throw new IqError.NOT_IMPLEMENTED("can't process multiple content nodes"); + } + StanzaNode content = contents[0]; + string? name = content.get_attribute("name"); + StanzaNode description = get_single_node_anyns(content, "description"); + StanzaNode transport_node = get_single_node_anyns(content, "transport"); + if (name == null) { + throw new IqError.BAD_REQUEST("missing name"); + } + + Transport? transport = get_transport(transport_node.ns_uri); + TransportParameters? transport_params = null; + if (transport != null) { + transport_params = transport.parse_transport_parameters(transport_node); + } else { + // terminate the session below + } + + ContentType? content_type = get_content_type(description.ns_uri); + if (content_type == null) { + // TODO(hrxi): how do we signal an unknown content type? + throw new IqError.NOT_IMPLEMENTED("unknown content type"); + } + ContentParameters content_params = content_type.parse_content_parameters(description); + + TransportType type = content_type.content_type_transport_type(); + Session session = new Session.initiate_received(sid, type, transport_params, iq.from, name); + stream.get_flag(Flag.IDENTITY).add_session(session); + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq)); + + if (transport == null || transport.transport_type() != type) { + StanzaNode reason = new StanzaNode.build("reason", NS_URI) + .put_node(new StanzaNode.build("unsupported-transports", NS_URI)); + session.terminate(stream, reason); + return; + } + + content_params.on_session_initiate(stream, session); + } + + public void on_iq_set(XmppStream stream, Iq.Stanza iq) { + try { + handle_iq_set(stream, iq); + } catch (IqError e) { + send_iq_error(e, stream, iq); + } + } + + public void handle_iq_set(XmppStream stream, Iq.Stanza iq) throws IqError { + StanzaNode? jingle = iq.stanza.get_subnode("jingle", NS_URI); + string? sid = jingle != null ? jingle.get_attribute("sid") : null; + string? action = jingle != null ? jingle.get_attribute("action") : null; + if (jingle == null || sid == null || action == null) { + throw new IqError.BAD_REQUEST("missing jingle node, sid or action"); + } + Session? session = stream.get_flag(Flag.IDENTITY).get_session(sid); + if (action == "session-initiate") { + if (session != null) { + // TODO(hrxi): Info leak if other clients use predictable session IDs? + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.build(ErrorStanza.TYPE_MODIFY, ErrorStanza.CONDITION_CONFLICT, "session ID already in use", null))); + return; + } + handle_session_initiate(stream, sid, jingle, iq); + return; + } + if (session == null) { + StanzaNode unknown_session = new StanzaNode.build("unknown-session", ERROR_NS_URI).add_self_xmlns(); + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.error(iq, new ErrorStanza.item_not_found(unknown_session))); + return; + } + session.handle_iq_set(stream, action, jingle, iq); + } + + public override string get_ns() { return NS_URI; } + public override string get_id() { return IDENTITY.id; } +} + +public enum TransportType { + DATAGRAM, + STREAMING, +} + +public enum Senders { + BOTH, + INITIATOR, + NONE, + RESPONDER; + + public string to_string() { + switch (this) { + case BOTH: return "both"; + case INITIATOR: return "initiator"; + case NONE: return "none"; + case RESPONDER: return "responder"; + } + assert_not_reached(); + } +} + +public interface Transport : Object { + public abstract string transport_ns_uri(); + public abstract bool is_transport_available(XmppStream stream, Jid full_jid); + public abstract TransportType transport_type(); + public abstract TransportParameters create_transport_parameters(); + public abstract TransportParameters parse_transport_parameters(StanzaNode transport) throws IqError; +} + +public interface TransportParameters : Object { + public abstract string transport_ns_uri(); + public abstract StanzaNode to_transport_stanza_node(); + public abstract void update_transport(StanzaNode transport) throws IqError; + public abstract IOStream create_transport_connection(XmppStream stream, Jid peer_full_jid, Role role); +} + +public enum Role { + INITIATOR, + RESPONDER; + + public string to_string() { + switch (this) { + case INITIATOR: return "initiator"; + case RESPONDER: return "responder"; + } + assert_not_reached(); + } +} + +public interface ContentType : Object { + public abstract string content_type_ns_uri(); + public abstract TransportType content_type_transport_type(); + public abstract ContentParameters parse_content_parameters(StanzaNode description) throws IqError; +} + +public interface ContentParameters : Object { + public abstract void on_session_initiate(XmppStream stream, Session session); +} + + +public class Session { + // INITIATE_SENT -> ACTIVE -> ENDED + // INITIATE_RECEIVED -> ACTIVE -> ENDED + public enum State { + INITIATE_SENT, + INITIATE_RECEIVED, + ACTIVE, + ENDED, + } + + public State state { get; private set; } + + public string sid { get; private set; } + public Type type_ { get; private set; } + public Jid peer_full_jid { get; private set; } + public string content_name { get; private set; } + + // INITIATE_SENT | INITIATE_RECEIVED + TransportParameters? transport = null; + + // ACTIVE + public IOStream? conn { get; private set; } + + // Only interesting in INITIATE_SENT. + // Signals that the session has been accepted by the peer. + public signal void accepted(XmppStream stream); + + public Session.initiate_sent(string sid, Type type, TransportParameters transport, Jid peer_full_jid, string content_name) { + this.state = INITIATE_SENT; + this.sid = sid; + this.type_ = type; + this.peer_full_jid = peer_full_jid; + this.content_name = content_name; + this.transport = transport; + this.conn = null; + } + + public Session.initiate_received(string sid, Type type, TransportParameters? transport, Jid peer_full_jid, string content_name) { + this.state = INITIATE_RECEIVED; + this.sid = sid; + this.type_ = type; + this.peer_full_jid = peer_full_jid; + this.content_name = content_name; + this.transport = transport; + this.conn = null; + } + + public void handle_iq_set(XmppStream stream, string action, StanzaNode jingle, Iq.Stanza iq) throws IqError { + switch (action) { + case "session-accept": + if (state != INITIATE_SENT) { + throw new IqError.OUT_OF_ORDER("got session-accept while not waiting for one"); + } + handle_session_accept(stream, jingle, iq); + break; + case "session-terminate": + handle_session_terminate(stream, jingle, iq); + break; + case "content-accept": + case "content-add": + case "content-modify": + case "content-reject": + case "content-remove": + case "security-info": + case "transport-accept": + case "transport-info": + case "transport-reject": + case "transport-replace": + throw new IqError.NOT_IMPLEMENTED(@"$(action) is not implemented"); + default: + throw new IqError.BAD_REQUEST("invalid action"); + } + } + void handle_session_accept(XmppStream stream, StanzaNode jingle, Iq.Stanza iq) throws IqError { + string? responder_str = jingle.get_attribute("responder"); + Jid responder; + if (responder_str != null) { + responder = Jid.parse(responder_str) ?? iq.from; + } else { + responder = iq.from; // TODO(hrxi): and above, can we assume iq.from != null + // TODO(hrxi): more sanity checking, perhaps replace who we're talking to + } + if (!responder.is_full()) { + throw new IqError.BAD_REQUEST("invalid responder JID"); + } + Gee.List<StanzaNode> contents = jingle.get_subnodes("content"); + if (contents.size == 0) { + // TODO(hrxi): here and below, should we terminate the session? + throw new IqError.BAD_REQUEST("missing content node"); + } + if (contents.size > 1) { + throw new IqError.NOT_IMPLEMENTED("can't process multiple content nodes"); + } + StanzaNode content = contents[0]; + StanzaNode description = get_single_node_anyns(content, "description"); + StanzaNode transport_node = get_single_node_anyns(content, "transport"); + if (transport_node.ns_uri != transport.transport_ns_uri()) { + throw new IqError.BAD_REQUEST("session-accept with unnegotiated transport method"); + } + transport.update_transport(transport_node); + conn = transport.create_transport_connection(stream, peer_full_jid, Role.INITIATOR); + transport = null; + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq)); + state = ACTIVE; + accepted(stream); + } + void handle_session_terminate(XmppStream stream, StanzaNode jingle, Iq.Stanza iq) throws IqError { + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, new Iq.Stanza.result(iq)); + // TODO(hrxi): also handle presence type=unavailable + } + + public void accept(XmppStream stream, StanzaNode description) { + if (state != INITIATE_RECEIVED) { + return; // TODO(hrxi): what to do? + } + StanzaNode jingle = new StanzaNode.build("jingle", NS_URI) + .add_self_xmlns() + .put_attribute("action", "session-accept") + .put_attribute("sid", sid) + .put_node(new StanzaNode.build("content", NS_URI) + .put_attribute("creator", "initiator") + .put_attribute("name", content_name) + .put_node(description) + .put_node(transport.to_transport_stanza_node()) + ); + Iq.Stanza iq = new Iq.Stanza.set(jingle) { to=peer_full_jid }; + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq); + + conn = transport.create_transport_connection(stream, peer_full_jid, Role.RESPONDER); + transport = null; + + state = ACTIVE; + } + + public void reject(XmppStream stream) { + if (state != INITIATE_RECEIVED) { + return; // TODO(hrxi): what to do? + } + StanzaNode reason = new StanzaNode.build("reason", NS_URI) + .put_node(new StanzaNode.build("decline", NS_URI)); + terminate(stream, reason); + } + + public void set_application_error(XmppStream stream, StanzaNode? application_reason = null) { + StanzaNode reason = new StanzaNode.build("reason", NS_URI) + .put_node(new StanzaNode.build("failed-application", NS_URI)); + if (application_reason != null) { + reason.put_node(application_reason); + } + terminate(stream, reason); + } + + public void close_connection(XmppStream stream) { + if (state != ACTIVE) { + return; // TODO(hrxi): what to do? + } + conn.close(); + } + + public void terminate(XmppStream stream, StanzaNode reason) { + if (state != INITIATE_SENT && state != INITIATE_RECEIVED && state != ACTIVE) { + // TODO(hrxi): what to do? + return; + } + if (state == ACTIVE) { + conn.close(); + } + + StanzaNode jingle = new StanzaNode.build("jingle", NS_URI) + .add_self_xmlns() + .put_attribute("action", "session-terminate") + .put_attribute("sid", sid) + .put_node(reason); + Iq.Stanza iq = new Iq.Stanza.set(jingle) { to=peer_full_jid }; + stream.get_module(Iq.Module.IDENTITY).send_iq(stream, iq); + + state = ENDED; + // Immediately remove the session from the open sessions as per the + // XEP, don't wait for confirmation. + stream.get_flag(Flag.IDENTITY).remove_session(sid); + } +} + +public class Flag : XmppStreamFlag { + public static FlagIdentity<Flag> IDENTITY = new FlagIdentity<Flag>(NS_URI, "jingle"); + + private HashMap<string, Session> sessions = new HashMap<string, Session>(); + + public void add_session(Session session) { + sessions[session.sid] = session; + } + public Session? get_session(string sid) { + return sessions.has_key(sid) ? sessions[sid] : null; + } + public void remove_session(string sid) { + sessions.unset(sid); + } + + public override string get_ns() { return NS_URI; } + public override string get_id() { return IDENTITY.id; } +} + +} diff --git a/xmpp-vala/src/module/xep/0234_jingle_file_transfer.vala b/xmpp-vala/src/module/xep/0234_jingle_file_transfer.vala new file mode 100644 index 00000000..57222bae --- /dev/null +++ b/xmpp-vala/src/module/xep/0234_jingle_file_transfer.vala @@ -0,0 +1,130 @@ +using Gee; +using Xmpp; +using Xmpp.Xep; + +namespace Xmpp.Xep.JingleFileTransfer { + +private const string NS_URI = "urn:xmpp:jingle:apps:file-transfer:5"; + +public class Module : Jingle.ContentType, XmppStreamModule { + public static Xmpp.ModuleIdentity<Module> IDENTITY = new Xmpp.ModuleIdentity<Module>(NS_URI, "0234_jingle_file_transfer"); + + public override void attach(XmppStream stream) { + stream.get_module(ServiceDiscovery.Module.IDENTITY).add_feature(stream, NS_URI); + stream.get_module(Jingle.Module.IDENTITY).register_content_type(this); + } + public override void detach(XmppStream stream) { } + + public string content_type_ns_uri() { + return NS_URI; + } + public Jingle.TransportType content_type_transport_type() { + return Jingle.TransportType.STREAMING; + } + public Jingle.ContentParameters parse_content_parameters(StanzaNode description) throws Jingle.IqError { + return Parameters.parse(this, description); + } + + public signal void file_incoming(XmppStream stream, FileTransfer file_transfer); + + public bool is_available(XmppStream stream, Jid full_jid) { + bool? has_feature = stream.get_flag(ServiceDiscovery.Flag.IDENTITY).has_entity_feature(full_jid, NS_URI); + if (has_feature == null || !(!)has_feature) { + return false; + } + return stream.get_module(Jingle.Module.IDENTITY).is_available(stream, Jingle.TransportType.STREAMING, full_jid); + } + + public async void offer_file_stream(XmppStream stream, Jid receiver_full_jid, InputStream input_stream, string basename, int64 size) throws IOError { + StanzaNode description = new StanzaNode.build("description", NS_URI) + .add_self_xmlns() + .put_node(new StanzaNode.build("file", NS_URI) + .put_node(new StanzaNode.build("name", NS_URI).put_node(new StanzaNode.text(basename))) + .put_node(new StanzaNode.build("size", NS_URI).put_node(new StanzaNode.text(size.to_string())))); + // TODO(hrxi): Add the mandatory hash field + + Jingle.Session session = stream.get_module(Jingle.Module.IDENTITY) + .create_session(stream, Jingle.TransportType.STREAMING, receiver_full_jid, Jingle.Senders.INITIATOR, "a-file-offer", description); // TODO(hrxi): Why "a-file-offer"? + + SourceFunc callback = offer_file_stream.callback; + session.accepted.connect((stream) => { + session.conn.input_stream.close(); + Idle.add((owned) callback); + }); + yield; + + // TODO(hrxi): catch errors + yield session.conn.output_stream.splice_async(input_stream, OutputStreamSpliceFlags.CLOSE_SOURCE|OutputStreamSpliceFlags.CLOSE_TARGET); + } + + public override string get_ns() { return NS_URI; } + public override string get_id() { return IDENTITY.id; } +} + +public class Parameters : Jingle.ContentParameters, Object { + Module parent; + string? media_type; + public string? name { get; private set; } + public int64 size { get; private set; } + public StanzaNode original_description { get; private set; } + public Parameters(Module parent, StanzaNode original_description, string? media_type, string? name, int64? size) { + this.parent = parent; + this.original_description = original_description; + this.media_type = media_type; + this.name = name; + this.size = size; + } + public static Parameters parse(Module parent, StanzaNode description) throws Jingle.IqError { + Gee.List<StanzaNode> files = description.get_subnodes("file", NS_URI); + if (files.size != 1) { + throw new Jingle.IqError.BAD_REQUEST("there needs to be exactly one file node"); + } + StanzaNode file = files[0]; + StanzaNode? media_type_node = file.get_subnode("media-type", NS_URI); + StanzaNode? name_node = file.get_subnode("name", NS_URI); + StanzaNode? size_node = file.get_subnode("size", NS_URI); + string? media_type = media_type_node != null ? media_type_node.get_string_content() : null; + string? name = name_node != null ? name_node.get_string_content() : null; + string? size_raw = size_node != null ? size_node.get_string_content() : null; + // TODO(hrxi): For some reason, the ?:-expression does not work due to a type error. + //int64? size = size_raw != null ? int64.parse(size_raw) : null; // TODO(hrxi): this has no error handling + int64 size = -1; + if (size_raw != null) { + size = int64.parse(size_raw); + if (size < 0) { + throw new Jingle.IqError.BAD_REQUEST("negative file size is invalid"); + } + } + + return new Parameters(parent, description, media_type, name, size); + } + void on_session_initiate(XmppStream stream, Jingle.Session session) { + parent.file_incoming(stream, new FileTransfer(session, this)); + } +} + +public class FileTransfer : Object { + Jingle.Session session; + Parameters parameters; + + public Jid peer { get { return session.peer_full_jid; } } + public string? file_name { get { return parameters.name; } } + public int64 size { get { return parameters.size; } } + + public InputStream? stream { get { return session.conn != null ? session.conn.input_stream : null; } } + + public FileTransfer(Jingle.Session session, Parameters parameters) { + this.session = session; + this.parameters = parameters; + } + + public void accept(XmppStream stream) { + session.accept(stream, parameters.original_description); + session.conn.output_stream.close(); + } + public void reject(XmppStream stream) { + session.reject(stream); + } +} + +} diff --git a/xmpp-vala/src/module/xep/0261_jingle_in_band_bytestreams.vala b/xmpp-vala/src/module/xep/0261_jingle_in_band_bytestreams.vala new file mode 100644 index 00000000..dc2e8d7c --- /dev/null +++ b/xmpp-vala/src/module/xep/0261_jingle_in_band_bytestreams.vala @@ -0,0 +1,77 @@ +using Xmpp; +using Xmpp.Xep; + +namespace Xmpp.Xep.JingleInBandBytestreams { + +private const string NS_URI = "urn:xmpp:jingle:transports:ibb:1"; +private const int DEFAULT_BLOCKSIZE = 4096; +private const int MAX_BLOCKSIZE = 65535; + +public class Module : Jingle.Transport, XmppStreamModule { + public static Xmpp.ModuleIdentity<Module> IDENTITY = new Xmpp.ModuleIdentity<Module>(NS_URI, "0261_jingle_in_band_bytestreams"); + + public override void attach(XmppStream stream) { + stream.get_module(Jingle.Module.IDENTITY).register_transport(this); + stream.get_module(ServiceDiscovery.Module.IDENTITY).add_feature(stream, NS_URI); + } + public override void detach(XmppStream stream) { } + + public override string get_ns() { return NS_URI; } + public override string get_id() { return IDENTITY.id; } + + public bool is_transport_available(XmppStream stream, Jid full_jid) { + bool? result = stream.get_flag(ServiceDiscovery.Flag.IDENTITY).has_entity_feature(full_jid, NS_URI); + return result != null && result; + } + + public string transport_ns_uri() { + return NS_URI; + } + public Jingle.TransportType transport_type() { + return Jingle.TransportType.STREAMING; + } + public Jingle.TransportParameters create_transport_parameters() { + return new Parameters(random_uuid(), DEFAULT_BLOCKSIZE); + } + public Jingle.TransportParameters parse_transport_parameters(StanzaNode transport) throws Jingle.IqError { + return Parameters.parse(transport); + } +} + +class Parameters : Jingle.TransportParameters, Object { + public string sid { get; private set; } + public int block_size { get; private set; } + public Parameters(string sid, int block_size) { + this.sid = sid; + this.block_size = block_size; + } + public static Parameters parse(StanzaNode transport) throws Jingle.IqError { + string? sid = transport.get_attribute("sid"); + int block_size = transport.get_attribute_int("block-size"); + if (sid == null || block_size <= 0 || block_size > MAX_BLOCKSIZE) { + throw new Jingle.IqError.BAD_REQUEST("missing or invalid sid or blocksize"); + } + return new Parameters(sid, block_size); + } + public string transport_ns_uri() { + return NS_URI; + } + public StanzaNode to_transport_stanza_node() { + return new StanzaNode.build("transport", NS_URI) + .add_self_xmlns() + .put_attribute("block-size", block_size.to_string()) + .put_attribute("sid", sid); + } + public void update_transport(StanzaNode transport) throws Jingle.IqError { + Parameters other = Parameters.parse(transport); + if (other.sid != sid || other.block_size > block_size) { + throw new Jingle.IqError.NOT_ACCEPTABLE("invalid IBB sid or block_size"); + } + block_size = other.block_size; + } + public IOStream create_transport_connection(XmppStream stream, Jid peer_full_jid, Jingle.Role role) { + return InBandBytestreams.Connection.create(stream, peer_full_jid, sid, block_size, role == Jingle.Role.INITIATOR); + } +} + +} |