2 * Unix SMB/CIFS implementation.
3 * RPC client transport over tstream
4 * Copyright (C) Simo Sorce 2010
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License as published by
8 * the Free Software Foundation; either version 3 of the License, or
9 * (at your option) any later version.
11 * This program is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 * GNU General Public License for more details.
16 * You should have received a copy of the GNU General Public License
17 * along with this program; if not, see <http://www.gnu.org/licenses/>.
21 #include "../lib/util/tevent_ntstatus.h"
22 #include "rpc_client/rpc_transport.h"
23 #include "lib/tsocket/tsocket.h"
24 #include "libsmb/cli_np_tstream.h"
28 #define DBGC_CLASS DBGC_RPC_CLI
30 struct rpc_tstream_state {
31 struct tstream_context *stream;
32 struct tevent_queue *read_queue;
33 struct tevent_queue *write_queue;
37 static void rpc_tstream_disconnect(struct rpc_tstream_state *s)
39 TALLOC_FREE(s->stream);
42 static bool rpc_tstream_is_connected(void *priv)
44 struct rpc_tstream_state *transp =
45 talloc_get_type_abort(priv, struct rpc_tstream_state);
48 if (!transp->stream) {
52 if (!tstream_is_cli_np(transp->stream)) {
56 ret = tstream_pending_bytes(transp->stream);
64 static unsigned int rpc_tstream_set_timeout(void *priv, unsigned int timeout)
66 struct rpc_tstream_state *transp =
67 talloc_get_type_abort(priv, struct rpc_tstream_state);
71 ok = rpc_tstream_is_connected(transp);
76 if (tstream_is_cli_np(transp->stream)) {
77 transp->timeout = timeout;
78 return tstream_cli_np_set_timeout(transp->stream, timeout);
81 orig_timeout = transp->timeout;
83 transp->timeout = timeout;
88 struct rpc_tstream_next_vector_state {
95 static void rpc_tstream_next_vector_init(
96 struct rpc_tstream_next_vector_state *s,
97 uint8_t *buf, size_t len)
102 s->len = MIN(len, UINT16_MAX);
105 static int rpc_tstream_next_vector(struct tstream_context *stream,
108 struct iovec **_vector,
111 struct rpc_tstream_next_vector_state *state =
112 (struct rpc_tstream_next_vector_state *)private_data;
113 struct iovec *vector;
117 if (state->ofs == state->len) {
123 pending = tstream_pending_bytes(stream);
128 if (pending == 0 && state->ofs != 0) {
129 /* return a short read */
136 /* we want at least one byte and recheck again */
139 size_t missing = state->len - state->ofs;
140 if (pending > missing) {
141 /* there's more available */
142 state->remaining = pending - missing;
145 /* read what we can get and recheck in the next cycle */
150 vector = talloc_array(mem_ctx, struct iovec, 1);
155 vector[0].iov_base = state->buf + state->ofs;
156 vector[0].iov_len = wanted;
158 state->ofs += wanted;
165 struct rpc_tstream_read_state {
166 struct rpc_tstream_state *transp;
167 struct rpc_tstream_next_vector_state next_vector;
171 static void rpc_tstream_read_done(struct tevent_req *subreq);
173 static struct tevent_req *rpc_tstream_read_send(TALLOC_CTX *mem_ctx,
174 struct event_context *ev,
175 uint8_t *data, size_t size,
178 struct rpc_tstream_state *transp =
179 talloc_get_type_abort(priv, struct rpc_tstream_state);
180 struct tevent_req *req, *subreq;
181 struct rpc_tstream_read_state *state;
182 struct timeval endtime;
184 req = tevent_req_create(mem_ctx, &state, struct rpc_tstream_read_state);
188 if (!rpc_tstream_is_connected(transp)) {
189 tevent_req_nterror(req, NT_STATUS_CONNECTION_INVALID);
190 return tevent_req_post(req, ev);
192 state->transp = transp;
193 rpc_tstream_next_vector_init(&state->next_vector, data, size);
195 subreq = tstream_readv_pdu_queue_send(state, ev,
198 rpc_tstream_next_vector,
199 &state->next_vector);
200 if (subreq == NULL) {
201 tevent_req_nterror(req, NT_STATUS_NO_MEMORY);
202 return tevent_req_post(req, ev);
205 endtime = timeval_current_ofs_msec(transp->timeout);
206 if (!tevent_req_set_endtime(subreq, ev, endtime)) {
210 tevent_req_set_callback(subreq, rpc_tstream_read_done, req);
217 static void rpc_tstream_read_done(struct tevent_req *subreq)
219 struct tevent_req *req =
220 tevent_req_callback_data(subreq, struct tevent_req);
221 struct rpc_tstream_read_state *state =
222 tevent_req_data(req, struct rpc_tstream_read_state);
225 state->nread = tstream_readv_pdu_queue_recv(subreq, &err);
227 if (state->nread < 0) {
228 rpc_tstream_disconnect(state->transp);
229 tevent_req_nterror(req, map_nt_error_from_unix(err));
232 tevent_req_done(req);
235 static NTSTATUS rpc_tstream_read_recv(struct tevent_req *req, ssize_t *size)
237 struct rpc_tstream_read_state *state = tevent_req_data(
238 req, struct rpc_tstream_read_state);
241 if (tevent_req_is_nterror(req, &status)) {
244 *size = state->nread;
248 struct rpc_tstream_write_state {
249 struct event_context *ev;
250 struct rpc_tstream_state *transp;
255 static void rpc_tstream_write_done(struct tevent_req *subreq);
257 static struct tevent_req *rpc_tstream_write_send(TALLOC_CTX *mem_ctx,
258 struct event_context *ev,
259 const uint8_t *data, size_t size,
262 struct rpc_tstream_state *transp =
263 talloc_get_type_abort(priv, struct rpc_tstream_state);
264 struct tevent_req *req, *subreq;
265 struct rpc_tstream_write_state *state;
266 struct timeval endtime;
268 req = tevent_req_create(mem_ctx, &state, struct rpc_tstream_write_state);
272 if (!rpc_tstream_is_connected(transp)) {
273 tevent_req_nterror(req, NT_STATUS_CONNECTION_INVALID);
274 return tevent_req_post(req, ev);
277 state->transp = transp;
278 state->iov.iov_base = discard_const_p(void *, data);
279 state->iov.iov_len = size;
281 subreq = tstream_writev_queue_send(state, ev,
285 if (subreq == NULL) {
289 endtime = timeval_current_ofs_msec(transp->timeout);
290 if (!tevent_req_set_endtime(subreq, ev, endtime)) {
294 tevent_req_set_callback(subreq, rpc_tstream_write_done, req);
301 static void rpc_tstream_write_done(struct tevent_req *subreq)
303 struct tevent_req *req =
304 tevent_req_callback_data(subreq, struct tevent_req);
305 struct rpc_tstream_write_state *state =
306 tevent_req_data(req, struct rpc_tstream_write_state);
309 state->nwritten = tstream_writev_queue_recv(subreq, &err);
311 if (state->nwritten < 0) {
312 rpc_tstream_disconnect(state->transp);
313 tevent_req_nterror(req, map_nt_error_from_unix(err));
316 tevent_req_done(req);
319 static NTSTATUS rpc_tstream_write_recv(struct tevent_req *req, ssize_t *sent)
321 struct rpc_tstream_write_state *state =
322 tevent_req_data(req, struct rpc_tstream_write_state);
325 if (tevent_req_is_nterror(req, &status)) {
328 *sent = state->nwritten;
332 struct rpc_tstream_trans_state {
333 struct tevent_context *ev;
334 struct rpc_tstream_state *transp;
336 uint32_t max_rdata_len;
340 static void rpc_tstream_trans_writev(struct tevent_req *subreq);
341 static void rpc_tstream_trans_readv_pdu(struct tevent_req *subreq);
343 static int rpc_tstream_trans_next_vector(struct tstream_context *stream,
346 struct iovec **_vector,
349 static struct tevent_req *rpc_tstream_trans_send(TALLOC_CTX *mem_ctx,
350 struct tevent_context *ev,
351 uint8_t *data, size_t data_len,
352 uint32_t max_rdata_len,
355 struct rpc_tstream_state *transp =
356 talloc_get_type_abort(priv, struct rpc_tstream_state);
357 struct tevent_req *req, *subreq;
358 struct rpc_tstream_trans_state *state;
359 struct timeval endtime;
361 req = tevent_req_create(mem_ctx, &state,
362 struct rpc_tstream_trans_state);
367 if (!rpc_tstream_is_connected(transp)) {
368 tevent_req_nterror(req, NT_STATUS_CONNECTION_INVALID);
369 return tevent_req_post(req, ev);
372 state->transp = transp;
373 state->req.iov_len = data_len;
374 state->req.iov_base = discard_const_p(void *, data);
375 state->max_rdata_len = max_rdata_len;
377 endtime = timeval_current_ofs_msec(transp->timeout);
379 subreq = tstream_writev_queue_send(state, ev,
383 if (tevent_req_nomem(subreq, req)) {
384 return tevent_req_post(req, ev);
386 if (!tevent_req_set_endtime(subreq, ev, endtime)) {
387 return tevent_req_post(req, ev);
389 tevent_req_set_callback(subreq, rpc_tstream_trans_writev, req);
391 if (tstream_is_cli_np(transp->stream)) {
392 tstream_cli_np_use_trans(transp->stream);
395 subreq = tstream_readv_pdu_queue_send(state, ev,
398 rpc_tstream_trans_next_vector,
400 if (tevent_req_nomem(subreq, req)) {
401 return tevent_req_post(req, ev);
403 if (!tevent_req_set_endtime(subreq, ev, endtime)) {
404 return tevent_req_post(req, ev);
406 tevent_req_set_callback(subreq, rpc_tstream_trans_readv_pdu, req);
411 static void rpc_tstream_trans_writev(struct tevent_req *subreq)
413 struct tevent_req *req =
414 tevent_req_callback_data(subreq,
416 struct rpc_tstream_trans_state *state =
418 struct rpc_tstream_trans_state);
422 ret = tstream_writev_queue_recv(subreq, &err);
425 rpc_tstream_disconnect(state->transp);
426 tevent_req_nterror(req, map_nt_error_from_unix(err));
431 static int rpc_tstream_trans_next_vector(struct tstream_context *stream,
434 struct iovec **_vector,
437 struct rpc_tstream_trans_state *state =
438 talloc_get_type_abort(private_data,
439 struct rpc_tstream_trans_state);
440 struct iovec *vector;
442 if (state->max_rdata_len == state->rep.iov_len) {
448 state->rep.iov_base = talloc_array(state, uint8_t,
449 state->max_rdata_len);
450 if (state->rep.iov_base == NULL) {
453 state->rep.iov_len = state->max_rdata_len;
455 vector = talloc_array(mem_ctx, struct iovec, 1);
460 vector[0] = state->rep;
467 static void rpc_tstream_trans_readv_pdu(struct tevent_req *subreq)
469 struct tevent_req *req =
470 tevent_req_callback_data(subreq,
472 struct rpc_tstream_trans_state *state =
474 struct rpc_tstream_trans_state);
478 ret = tstream_readv_pdu_queue_recv(subreq, &err);
481 rpc_tstream_disconnect(state->transp);
482 tevent_req_nterror(req, map_nt_error_from_unix(err));
486 tevent_req_done(req);
489 static NTSTATUS rpc_tstream_trans_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
490 uint8_t **prdata, uint32_t *prdata_len)
492 struct rpc_tstream_trans_state *state =
494 struct rpc_tstream_trans_state);
497 if (tevent_req_is_nterror(req, &status)) {
501 *prdata = (uint8_t *)talloc_move(mem_ctx, &state->rep.iov_base);
502 *prdata_len = state->rep.iov_len;
507 * @brief Initialize a tstream transport facility
508 * NOTE: this function will talloc_steal, the stream and the queues.
510 * @param mem_ctx - memory context used to allocate the transport
511 * @param stream - a ready to use tstream
512 * @param presult - the transport structure
514 * @return - a NT Status error code.
516 NTSTATUS rpc_transport_tstream_init(TALLOC_CTX *mem_ctx,
517 struct tstream_context **stream,
518 struct rpc_cli_transport **presult)
520 struct rpc_cli_transport *result;
521 struct rpc_tstream_state *state;
523 result = talloc(mem_ctx, struct rpc_cli_transport);
524 if (result == NULL) {
525 return NT_STATUS_NO_MEMORY;
527 state = talloc(result, struct rpc_tstream_state);
530 return NT_STATUS_NO_MEMORY;
532 result->priv = state;
534 state->read_queue = tevent_queue_create(state, "read_queue");
535 if (state->read_queue == NULL) {
537 return NT_STATUS_NO_MEMORY;
539 state->write_queue = tevent_queue_create(state, "write_queue");
540 if (state->write_queue == NULL) {
542 return NT_STATUS_NO_MEMORY;
545 state->stream = talloc_move(state, stream);
546 state->timeout = 10000; /* 10 seconds. */
548 if (tstream_is_cli_np(state->stream)) {
549 result->trans_send = rpc_tstream_trans_send;
550 result->trans_recv = rpc_tstream_trans_recv;
552 result->trans_send = NULL;
553 result->trans_recv = NULL;
555 result->write_send = rpc_tstream_write_send;
556 result->write_recv = rpc_tstream_write_recv;
557 result->read_send = rpc_tstream_read_send;
558 result->read_recv = rpc_tstream_read_recv;
559 result->is_connected = rpc_tstream_is_connected;
560 result->set_timeout = rpc_tstream_set_timeout;
566 struct cli_state *rpc_pipe_np_smb_conn(struct rpc_pipe_client *p)
568 struct rpc_tstream_state *transp =
569 talloc_get_type_abort(p->transport->priv,
570 struct rpc_tstream_state);
573 ok = rpccli_is_connected(p);
578 if (!tstream_is_cli_np(transp->stream)) {
582 return tstream_cli_np_get_cli_state(transp->stream);