src/protocol-native.c

   1 #include <string.h>
   2 #include <stdio.h>
   3 #include <assert.h>
   4 #include <stdlib.h>
   5
   6 #include "protocol-native.h"
   7 #include "protocol-native-spec.h"
   8 #include "packet.h"
   9 #include "client.h"
  10 #include "sourceoutput.h"
  11 #include "sinkinput.h"
  12 #include "pstream.h"
  13 #include "tagstruct.h"
  14 #include "pdispatch.h"
  15 #include "pstream-util.h"
  16 #include "authkey.h"
  17 #include "namereg.h"
  18
  19 struct connection;
  20 struct pa_protocol_native;
  21
  22 struct record_stream {
  23     struct connection *connection;
  24     uint32_t index;
  25     struct pa_source_output *source_output;
  26     struct pa_memblockq *memblockq;
  27 };
  28
  29 struct playback_stream {
  30     struct connection *connection;
  31     uint32_t index;
  32     struct pa_sink_input *sink_input;
  33     struct pa_memblockq *memblockq;
  34     size_t requested_bytes;
  35     int drain_request;
  36     uint32_t drain_tag;
  37 };
  38
  39 struct connection {
  40     int authorized;
  41     struct pa_protocol_native *protocol;
  42     struct pa_client *client;
  43     struct pa_pstream *pstream;
  44     struct pa_pdispatch *pdispatch;
  45     struct pa_idxset *record_streams, *playback_streams;
  46 };
  47
  48 struct pa_protocol_native {
  49     int public;
  50     struct pa_core *core;
  51     struct pa_socket_server *server;
  52     struct pa_idxset *connections;
  53     uint8_t auth_cookie[PA_NATIVE_COOKIE_LENGTH];
  54 };
  55
  56 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk);
  57 static void sink_input_drop_cb(struct pa_sink_input *i, size_t length);
  58 static void sink_input_kill_cb(struct pa_sink_input *i);
  59 static uint32_t sink_input_get_latency_cb(struct pa_sink_input *i);
  60
  61 static void request_bytes(struct playback_stream*s);
  62
  63 static void command_exit(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  64 static void command_create_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  65 static void command_delete_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  66 static void command_drain_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  67 static void command_auth(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  68 static void command_set_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  69 static void command_lookup(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  70
  71 static const struct pa_pdispatch_command command_table[PA_COMMAND_MAX] = {
  72     [PA_COMMAND_ERROR] = { NULL },
  73     [PA_COMMAND_TIMEOUT] = { NULL },
  74     [PA_COMMAND_REPLY] = { NULL },
  75     [PA_COMMAND_CREATE_PLAYBACK_STREAM] = { command_create_playback_stream },
  76     [PA_COMMAND_DELETE_PLAYBACK_STREAM] = { command_delete_playback_stream },
  77     [PA_COMMAND_DRAIN_PLAYBACK_STREAM] = { command_drain_playback_stream },
  78     [PA_COMMAND_CREATE_RECORD_STREAM] = { NULL },
  79     [PA_COMMAND_DELETE_RECORD_STREAM] = { NULL },
  80     [PA_COMMAND_AUTH] = { command_auth },
  81     [PA_COMMAND_REQUEST] = { NULL },
  82     [PA_COMMAND_EXIT] = { command_exit },
  83     [PA_COMMAND_SET_NAME] = { command_set_name },
  84     [PA_COMMAND_LOOKUP_SINK] = { command_lookup },
  85     [PA_COMMAND_LOOKUP_SOURCE] = { command_lookup },
  86 };
  87
  88 /* structure management */
  89
  90 static void record_stream_free(struct record_stream* r) {
  91     assert(r && r->connection);
  92
  93     pa_idxset_remove_by_data(r->connection->record_streams, r, NULL);
  94     pa_source_output_free(r->source_output);
  95     pa_memblockq_free(r->memblockq);
  96     free(r);
  97 }
  98
  99 static struct playback_stream* playback_stream_new(struct connection *c, struct pa_sink *sink, struct pa_sample_spec *ss, const char *name,
 100                                                    size_t maxlength,
 101                                                    size_t tlength,
 102                                                    size_t prebuf,
 103                                                    size_t minreq) {
 104     struct playback_stream *s;
 105     assert(c && sink && ss && name && maxlength);
 106
 107     s = malloc(sizeof(struct playback_stream));
 108     assert (s);
 109     s->connection = c;
 110
 111     s->sink_input = pa_sink_input_new(sink, name, ss);
 112     assert(s->sink_input);
 113     s->sink_input->peek = sink_input_peek_cb;
 114     s->sink_input->drop = sink_input_drop_cb;
 115     s->sink_input->kill = sink_input_kill_cb;
 116     s->sink_input->get_latency = sink_input_get_latency_cb;
 117     s->sink_input->userdata = s;
 118
 119     s->memblockq = pa_memblockq_new(maxlength, tlength, pa_sample_size(ss), prebuf, minreq);
 120     assert(s->memblockq);
 121
 122     s->requested_bytes = 0;
 123     s->drain_request = 0;
 124
 125     pa_idxset_put(c->playback_streams, s, &s->index);
 126     return s;
 127 }
 128
 129 static void playback_stream_free(struct playback_stream* p) {
 130     assert(p && p->connection);
 131
 132     if (p->drain_request)
 133         pa_pstream_send_error(p->connection->pstream, p->drain_tag, PA_ERROR_NOENTITY);
 134
 135     pa_idxset_remove_by_data(p->connection->playback_streams, p, NULL);
 136     pa_sink_input_free(p->sink_input);
 137     pa_memblockq_free(p->memblockq);
 138     free(p);
 139 }
 140
 141 static void connection_free(struct connection *c) {
 142     struct record_stream *r;
 143     struct playback_stream *p;
 144     assert(c && c->protocol);
 145
 146     pa_idxset_remove_by_data(c->protocol->connections, c, NULL);
 147     while ((r = pa_idxset_first(c->record_streams, NULL)))
 148         record_stream_free(r);
 149     pa_idxset_free(c->record_streams, NULL, NULL);
 150
 151     while ((p = pa_idxset_first(c->playback_streams, NULL)))
 152         playback_stream_free(p);
 153     pa_idxset_free(c->playback_streams, NULL, NULL);
 154
 155     pa_pdispatch_free(c->pdispatch);
 156     pa_pstream_free(c->pstream);
 157     pa_client_free(c->client);
 158     free(c);
 159 }
 160
 161 static void request_bytes(struct playback_stream *s) {
 162     struct pa_tagstruct *t;
 163     size_t l;
 164     assert(s);
 165
 166     if (!(l = pa_memblockq_missing(s->memblockq)))
 167         return;
 168
 169     if (l <= s->requested_bytes)
 170         return;
 171
 172     l -= s->requested_bytes;
 173
 174     if (l < pa_memblockq_get_minreq(s->memblockq))
 175         return;
 176
 177     s->requested_bytes += l;
 178
 179     t = pa_tagstruct_new(NULL, 0);
 180     assert(t);
 181     pa_tagstruct_putu32(t, PA_COMMAND_REQUEST);
 182     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 183     pa_tagstruct_putu32(t, s->index);
 184     pa_tagstruct_putu32(t, l);
 185     pa_pstream_send_tagstruct(s->connection->pstream, t);
 186
 187     /*fprintf(stderr, "Requesting %u bytes\n", l);*/
 188 }
 189
 190 /*** sinkinput callbacks ***/
 191
 192 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk) {
 193     struct playback_stream *s;
 194     assert(i && i->userdata && chunk);
 195     s = i->userdata;
 196
 197     if (pa_memblockq_peek(s->memblockq, chunk) < 0)
 198         return -1;
 199
 200     return 0;
 201 }
 202
 203 static void sink_input_drop_cb(struct pa_sink_input *i, size_t length) {
 204     struct playback_stream *s;
 205     assert(i && i->userdata && length);
 206     s = i->userdata;
 207
 208     pa_memblockq_drop(s->memblockq, length);
 209     request_bytes(s);
 210
 211     if (s->drain_request && !pa_memblockq_is_readable(s->memblockq)) {
 212         pa_pstream_send_simple_ack(s->connection->pstream, s->drain_tag);
 213         s->drain_request = 0;
 214     }
 215 }
 216
 217 static void sink_input_kill_cb(struct pa_sink_input *i) {
 218     struct playback_stream *s;
 219     assert(i && i->userdata);
 220     s = i->userdata;
 221
 222     playback_stream_free(s);
 223 }
 224
 225 static uint32_t sink_input_get_latency_cb(struct pa_sink_input *i) {
 226     struct playback_stream *s;
 227     assert(i && i->userdata);
 228     s = i->userdata;
 229
 230     return pa_samples_usec(pa_memblockq_get_length(s->memblockq), &s->sink_input->sample_spec);
 231 }
 232
 233 /*** pdispatch callbacks ***/
 234
 235 static void protocol_error(struct connection *c) {
 236     fprintf(stderr, __FILE__": protocol error, kicking client\n");
 237     connection_free(c);
 238 }
 239
 240 static void command_create_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 241     struct connection *c = userdata;
 242     struct playback_stream *s;
 243     size_t maxlength, tlength, prebuf, minreq;
 244     uint32_t sink_index;
 245     const char *name;
 246     struct pa_sample_spec ss;
 247     struct pa_tagstruct *reply;
 248     struct pa_sink *sink;
 249     assert(c && t && c->protocol && c->protocol->core);
 250
 251     if (pa_tagstruct_gets(t, &name) < 0 ||
 252         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 253         pa_tagstruct_getu32(t, &sink_index) < 0 ||
 254         pa_tagstruct_getu32(t, &maxlength) < 0 ||
 255         pa_tagstruct_getu32(t, &tlength) < 0 ||
 256         pa_tagstruct_getu32(t, &prebuf) < 0 ||
 257         pa_tagstruct_getu32(t, &minreq) < 0 ||
 258         !pa_tagstruct_eof(t)) {
 259         protocol_error(c);
 260         return;
 261     }
 262
 263     if (!c->authorized) {
 264         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 265         return;
 266     }
 267
 268     if (sink_index == (uint32_t) -1)
 269         sink = pa_sink_get_default(c->protocol->core);
 270     else
 271         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
 272
 273     if (!sink) {
 274         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 275         return;
 276     }
 277
 278     if (!(s = playback_stream_new(c, sink, &ss, name, maxlength, tlength, prebuf, minreq))) {
 279         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 280         return;
 281     }
 282
 283     reply = pa_tagstruct_new(NULL, 0);
 284     assert(reply);
 285     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 286     pa_tagstruct_putu32(reply, tag);
 287     pa_tagstruct_putu32(reply, s->index);
 288     assert(s->sink_input);
 289     pa_tagstruct_putu32(reply, s->sink_input->index);
 290     pa_pstream_send_tagstruct(c->pstream, reply);
 291     request_bytes(s);
 292 }
 293
 294 static void command_delete_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 295     struct connection *c = userdata;
 296     uint32_t channel;
 297     struct playback_stream *s;
 298     assert(c && t);
 299
 300     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 301         !pa_tagstruct_eof(t)) {
 302         protocol_error(c);
 303         return;
 304     }
 305
 306     if (!c->authorized) {
 307         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 308         return;
 309     }
 310
 311     if (!(s = pa_idxset_get_by_index(c->playback_streams, channel))) {
 312         pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 313         return;
 314     }
 315
 316     pa_pstream_send_simple_ack(c->pstream, tag);
 317 }
 318
 319 static void command_exit(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 320     struct connection *c = userdata;
 321     assert(c && t);
 322
 323     if (!pa_tagstruct_eof(t)) {
 324         protocol_error(c);
 325         return;
 326     }
 327
 328     if (!c->authorized) {
 329         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 330         return;
 331     }
 332
 333     assert(c->protocol && c->protocol->core && c->protocol->core->mainloop);
 334     c->protocol->core->mainloop->quit(c->protocol->core->mainloop, 0);
 335     pa_pstream_send_simple_ack(c->pstream, tag); /* nonsense */
 336     return;
 337 }
 338
 339 static void command_auth(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 340     struct connection *c = userdata;
 341     const void*cookie;
 342     assert(c && t);
 343
 344     if (pa_tagstruct_get_arbitrary(t, &cookie, PA_NATIVE_COOKIE_LENGTH) < 0 ||
 345         !pa_tagstruct_eof(t)) {
 346         protocol_error(c);
 347         return;
 348     }
 349
 350     if (memcmp(c->protocol->auth_cookie, cookie, PA_NATIVE_COOKIE_LENGTH) != 0) {
 351         fprintf(stderr, "protocol-native.c: Denied access to client with invalid authorization key.\n");
 352         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 353         return;
 354     }
 355
 356     c->authorized = 1;
 357     pa_pstream_send_simple_ack(c->pstream, tag);
 358     return;
 359 }
 360
 361 static void command_set_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 362     struct connection *c = userdata;
 363     const char *name;
 364     assert(c && t);
 365
 366     if (pa_tagstruct_gets(t, &name) < 0 ||
 367         !pa_tagstruct_eof(t)) {
 368         protocol_error(c);
 369         return;
 370     }
 371
 372     pa_client_rename(c->client, name);
 373     pa_pstream_send_simple_ack(c->pstream, tag);
 374     return;
 375 }
 376
 377 static void command_lookup(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 378     struct connection *c = userdata;
 379     const char *name;
 380     uint32_t index = PA_IDXSET_INVALID;
 381     assert(c && t);
 382
 383     if (pa_tagstruct_gets(t, &name) < 0 ||
 384         !pa_tagstruct_eof(t)) {
 385         protocol_error(c);
 386         return;
 387     }
 388
 389     if (!c->authorized) {
 390         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 391         return;
 392     }
 393
 394     if (command == PA_COMMAND_LOOKUP_SINK) {
 395         struct pa_sink *sink;
 396         if ((sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK)))
 397             index = sink->index;
 398     } else {
 399         struct pa_source *source;
 400         assert(command == PA_COMMAND_LOOKUP_SOURCE);
 401         if ((source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE)))
 402             index = source->index;
 403     }
 404
 405     if (index == PA_IDXSET_INVALID)
 406         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 407     else {
 408         struct pa_tagstruct *reply;
 409         reply = pa_tagstruct_new(NULL, 0);
 410         assert(reply);
 411         pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 412         pa_tagstruct_putu32(reply, tag);
 413         pa_tagstruct_putu32(reply, index);
 414         pa_pstream_send_tagstruct(c->pstream, reply);
 415     }
 416 }
 417
 418 static void command_drain_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 419     struct connection *c = userdata;
 420     uint32_t index;
 421     struct playback_stream *s;
 422     assert(c && t);
 423
 424     if (pa_tagstruct_getu32(t, &index) < 0 ||
 425         !pa_tagstruct_eof(t)) {
 426         protocol_error(c);
 427         return;
 428     }
 429
 430     if (!c->authorized) {
 431         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 432         return;
 433     }
 434
 435     if (!(s = pa_idxset_get_by_index(c->playback_streams, index))) {
 436         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 437         return;
 438     }
 439
 440     s->drain_request = 0;
 441
 442     if (!pa_memblockq_is_readable(s->memblockq))
 443         pa_pstream_send_simple_ack(c->pstream, tag);
 444     else {
 445         s->drain_request = 1;
 446         s->drain_tag = tag;
 447     }
 448 }
 449
 450 /*** pstream callbacks ***/
 451
 452 static void packet_callback(struct pa_pstream *p, struct pa_packet *packet, void *userdata) {
 453     struct connection *c = userdata;
 454     assert(p && packet && packet->data && c);
 455
 456     if (pa_pdispatch_run(c->pdispatch, packet, c) < 0) {
 457         fprintf(stderr, "protocol-native: invalid packet.\n");
 458         connection_free(c);
 459     }
 460 }
 461
 462 static void memblock_callback(struct pa_pstream *p, uint32_t channel, int32_t delta, struct pa_memchunk *chunk, void *userdata) {
 463     struct connection *c = userdata;
 464     struct playback_stream *stream;
 465     assert(p && chunk && userdata);
 466
 467     if (!(stream = pa_idxset_get_by_index(c->playback_streams, channel))) {
 468         fprintf(stderr, "protocol-native: client sent block for invalid stream.\n");
 469         connection_free(c);
 470         return;
 471     }
 472
 473     if (chunk->length >= stream->requested_bytes)
 474         stream->requested_bytes = 0;
 475     else
 476         stream->requested_bytes -= chunk->length;
 477
 478     pa_memblockq_push_align(stream->memblockq, chunk, delta);
 479     assert(stream->sink_input);
 480     pa_sink_notify(stream->sink_input->sink);
 481
 482     /*fprintf(stderr, "Recieved %u bytes.\n", chunk->length);*/
 483 }
 484
 485 static void die_callback(struct pa_pstream *p, void *userdata) {
 486     struct connection *c = userdata;
 487     assert(p && c);
 488     connection_free(c);
 489
 490     fprintf(stderr, "protocol-native: connection died.\n");
 491 }
 492
 493 /*** client callbacks ***/
 494
 495 static void client_kill_cb(struct pa_client *c) {
 496     assert(c && c->userdata);
 497     connection_free(c->userdata);
 498 }
 499
 500 /*** socket server callbacks ***/
 501
 502 static void on_connection(struct pa_socket_server*s, struct pa_iochannel *io, void *userdata) {
 503     struct pa_protocol_native *p = userdata;
 504     struct connection *c;
 505     assert(s && io && p);
 506
 507     c = malloc(sizeof(struct connection));
 508     assert(c);
 509     c->authorized = p->public;
 510     c->protocol = p;
 511     assert(p->core);
 512     c->client = pa_client_new(p->core, "NATIVE", "Client");
 513     assert(c->client);
 514     c->client->kill = client_kill_cb;
 515     c->client->userdata = c;
 516     c->pstream = pa_pstream_new(p->core->mainloop, io);
 517     assert(c->pstream);
 518
 519     pa_pstream_set_recieve_packet_callback(c->pstream, packet_callback, c);
 520     pa_pstream_set_recieve_memblock_callback(c->pstream, memblock_callback, c);
 521     pa_pstream_set_die_callback(c->pstream, die_callback, c);
 522
 523     c->pdispatch = pa_pdispatch_new(p->core->mainloop, command_table, PA_COMMAND_MAX);
 524     assert(c->pdispatch);
 525
 526     c->record_streams = pa_idxset_new(NULL, NULL);
 527     c->playback_streams = pa_idxset_new(NULL, NULL);
 528     assert(c->record_streams && c->playback_streams);
 529
 530     pa_idxset_put(p->connections, c, NULL);
 531 }
 532
 533 /*** module entry points ***/
 534
 535 struct pa_protocol_native* pa_protocol_native_new(struct pa_core *core, struct pa_socket_server *server) {
 536     struct pa_protocol_native *p;
 537     assert(core && server);
 538
 539     p = malloc(sizeof(struct pa_protocol_native));
 540     assert(p);
 541
 542     if (pa_authkey_load_from_home(PA_NATIVE_COOKIE_FILE, p->auth_cookie, sizeof(p->auth_cookie)) < 0) {
 543         free(p);
 544         return NULL;
 545     }
 546
 547     p->public = 1;
 548     p->server = server;
 549     p->core = core;
 550     p->connections = pa_idxset_new(NULL, NULL);
 551     assert(p->connections);
 552
 553     pa_socket_server_set_callback(p->server, on_connection, p);
 554
 555     return p;
 556 }
 557
 558 void pa_protocol_native_free(struct pa_protocol_native *p) {
 559     struct connection *c;
 560     assert(p);
 561
 562     while ((c = pa_idxset_first(p->connections, NULL)))
 563         connection_free(c);
 564     pa_idxset_free(p->connections, NULL, NULL);
 565     pa_socket_server_free(p->server);
 566     free(p);
 567 }