X-Git-Url: http://www.project-moonshot.org/gitweb/?a=blobdiff_plain;f=trp%2Ftrps.c;h=193faaa9f8b18ea0418e0f3cd29e4f0b059799f6;hb=6491f2a660c3183b210b968a939eaa18f1b44a09;hp=cbe039731a8a8ff4daa5bccca1f4770f3d39b539;hpb=e9332953189a29c0c52f703e4a3717b6141306d9;p=trust_router.git diff --git a/trp/trps.c b/trp/trps.c index cbe0397..193faaa 100644 --- a/trp/trps.c +++ b/trp/trps.c @@ -39,18 +39,20 @@ #include #include #include +#include // for nfds_t #include #include #include #include -#include +#include #include -#include +#include #include #include #include #include +#include static int trps_destructor(void *object) { @@ -163,8 +165,6 @@ void trps_set_sweep_interval(TRPS_INSTANCE *trps, unsigned int interval) void trps_set_ctable(TRPS_INSTANCE *trps, TR_COMM_TABLE *comm) { - if (trps->ctable!=NULL) - tr_comm_table_free(trps->ctable); trps->ctable=comm; } @@ -278,82 +278,6 @@ TRP_RC trps_send_msg(TRPS_INSTANCE *trps, TRP_PEER *peer, const char *msg) return rc; } -/* Listens on all interfaces. Returns number of sockets opened. Their - * descriptors are stored in *fd_out, which should point to space for - * up to max_fd of them. */ -static size_t trps_listen(TRPS_INSTANCE *trps, int port, int *fd_out, size_t max_fd) -{ - int rc = 0; - int conn = -1; - int optval=0; - struct addrinfo *ai=NULL; - struct addrinfo *ai_head=NULL; - struct addrinfo hints={.ai_flags=AI_PASSIVE, - .ai_family=AF_UNSPEC, - .ai_socktype=SOCK_STREAM, - .ai_protocol=IPPROTO_TCP}; - char *port_str=NULL; - size_t n_opened=0; - - port_str=talloc_asprintf(NULL, "%d", port); - if (port_str==NULL) { - tr_debug("trps_listen: unable to allocate port."); - return -1; - } - getaddrinfo(NULL, port_str, &hints, &ai_head); - talloc_free(port_str); - - for (ai=ai_head,n_opened=0; (ai!=NULL)&&(n_openedai_next) { - if (0 > (conn = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol))) { - tr_debug("trps_listen: unable to open socket."); - continue; - } - - optval=1; - if (0!=setsockopt(conn, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval))) - tr_debug("trps_listen: unable to set SO_REUSEADDR."); /* not fatal? */ - - if (ai->ai_family==AF_INET6) { - /* don't allow IPv4-mapped IPv6 addresses (per RFC4942, not sure - * if still relevant) */ - if (0!=setsockopt(conn, IPPROTO_IPV6, IPV6_V6ONLY, &optval, sizeof(optval))) { - tr_debug("trps_listen: unable to set IPV6_V6ONLY. Skipping interface."); - close(conn); - continue; - } - } - - rc=bind(conn, ai->ai_addr, ai->ai_addrlen); - if (rc<0) { - tr_debug("trps_listen: unable to bind to socket."); - close(conn); - continue; - } - - if (0>listen(conn, 512)) { - tr_debug("trps_listen: unable to listen on bound socket."); - close(conn); - continue; - } - - /* ok, this one worked. Save it */ - fd_out[n_opened++]=conn; - } - freeaddrinfo(ai_head); - - if (n_opened==0) { - tr_debug("trps_listen: no addresses available for listening."); - return -1; - } - - tr_debug("trps_listen: TRP Server listening on port %d on %d socket%s", - port, - n_opened, - (n_opened==1)?"":"s"); - - return n_opened; -} - /* get the currently selected route if available */ TRP_ROUTE *trps_get_route(TRPS_INSTANCE *trps, TR_NAME *comm, TR_NAME *realm, TR_NAME *peer) { @@ -412,7 +336,7 @@ static TRP_RC trps_read_message(TRPS_INSTANCE *trps, TRP_CONNECTION *conn, TR_MS tr_debug("trps_read_message: message received, %u bytes.", (unsigned) buflen); tr_debug("trps_read_message: %.*s", buflen, buf); - *msg=tr_msg_decode(buf, buflen); + *msg= tr_msg_decode(NULL, buf, buflen); free(buf); if (*msg==NULL) return TRP_NOPARSE; @@ -434,6 +358,8 @@ static TRP_RC trps_read_message(TRPS_INSTANCE *trps, TRP_CONNECTION *conn, TR_MS case TRP_UPDATE: trp_upd_set_peer(tr_msg_get_trp_upd(*msg), tr_dup_name(conn_peer)); trp_upd_set_next_hop(tr_msg_get_trp_upd(*msg), trp_peer_get_server(peer), 0); /* TODO: 0 should be the configured TID port */ + /* update provenance if necessary */ + trp_upd_add_to_provenance(tr_msg_get_trp_upd(*msg), trp_peer_get_label(peer)); break; case TRP_REQUEST: @@ -459,31 +385,32 @@ int trps_get_listener(TRPS_INSTANCE *trps, int *fd_out, size_t max_fd) { - size_t n_fd=0; - size_t ii=0; + nfds_t n_fd=0; + nfds_t ii=0; + + n_fd = tr_sock_listen_all(port, fd_out, max_fd); - n_fd=trps_listen(trps, port, fd_out, max_fd); - if (n_fd==0) - tr_debug("trps_get_listener: Error opening port %d."); + if (n_fd == 0) + tr_err("trps_get_listener: Error opening port %d."); else { /* opening port succeeded */ - tr_debug("trps_get_listener: Opened port %d.", port); + tr_info("trps_get_listener: Opened port %d.", port); /* make the sockets non-blocking */ for (ii=0; ii0) { + if (n_fd > 0) { /* store the caller's request handler & cookie */ trps->msg_handler = msg_handler; trps->auth_handler = auth_handler; @@ -492,7 +419,7 @@ int trps_get_listener(TRPS_INSTANCE *trps, trps->cookie = cookie; } - return n_fd; + return (int) n_fd; } TRP_RC trps_authorize_connection(TRPS_INSTANCE *trps, TRP_CONNECTION *conn) @@ -651,11 +578,12 @@ static int trps_check_feasibility(TRPS_INSTANCE *trps, TR_NAME *realm, TR_NAME * static struct timespec *trps_compute_expiry(TRPS_INSTANCE *trps, unsigned int interval, struct timespec *ts) { const unsigned int small_factor=3; /* how many intervals we wait before expiring */ - if (0!=clock_gettime(CLOCK_REALTIME, ts)) { + if (0!=clock_gettime(TRP_CLOCK, ts)) { tr_err("trps_compute_expiry: could not read realtime clock."); ts->tv_sec=0; ts->tv_nsec=0; } + tr_debug("trps_compute_expiry: tv_sec=%u, interval=%u, small_factor*interval=%u", ts->tv_sec, interval, small_factor*interval); ts->tv_sec += small_factor*interval; return ts; } @@ -922,7 +850,7 @@ static TRP_RC trps_handle_inforec_comm(TRPS_INSTANCE *trps, TRP_UPD *upd, TRP_IN } if (trps_name_in_provenance(our_peer_label, trp_inforec_get_provenance(rec))) - tr_debug("trps_handle_inforec_comm: rejecting community inforec to avoid loop."); + tr_debug("trps_handle_inforec_comm: rejecting community inforec to avoid provenance loop."); else { /* no loop occurring, accept the update */ comm=tr_comm_table_find_comm(trps->ctable, comm_id); @@ -1002,6 +930,56 @@ cleanup: return rc; } +/** + * Apply applicable TRP_INBOUND filters to an inforec. Rejects everything if peer has no filters. + * + * @param trps Active TRPS instance + * @param upd TRP_UPD that contains the inforec to filter + * @param rec Inforec to filter + * @return 1 if accepted by the filter, 0 otherwise + */ +static int trps_filter_inbound_inforec(TRPS_INSTANCE *trps, TRP_UPD *upd, TRP_INFOREC *rec) +{ + TRP_PEER *peer=NULL; + TR_NAME *peer_name=NULL; + TR_FILTER_ACTION action=TR_FILTER_ACTION_REJECT; + TR_FILTER_TARGET *target=NULL; + int retval=0; + + /* Look up the peer. For inbound messages, the peer is identified by its GSS name */ + peer_name=trp_upd_get_peer(upd); + peer=trps_get_peer_by_gssname(trps, peer_name); + if (peer==NULL) { + tr_err("trps_filter_inbound_inforec: received inforec from unknown peer (%.*s), rejecting.", + peer_name->len, + peer_name->buf); + return 0; + } + + /* tr_filter_apply() and tr_filter_set_get() handle null filter sets/filters by rejecting */ + target= tr_filter_target_trp_inforec(NULL, upd, rec); + if (target==NULL) { + /* TODO: signal that filtering failed. Until then, just filter everything and give an error message. */ + tr_crit("trps_filter_inbound_inforec: Unable to allocate filter target, cannot apply filter!"); + } + if ((target==NULL) + || (TR_FILTER_NO_MATCH==tr_filter_apply(target, + tr_filter_set_get(peer->filters, TR_FILTER_TYPE_TRP_INBOUND), + NULL, + &action)) + || (action!=TR_FILTER_ACTION_ACCEPT)) { + /* either the filter did not match or it matched a reject rule or allocating the target failed */ + retval=0; + } else + retval=1; + if (target!=NULL) + tr_filter_target_free(target); + + /* filter matched an accept rule */ + return retval; +} + + static TRP_RC trps_handle_update(TRPS_INSTANCE *trps, TRP_UPD *upd) { TRP_INFOREC *rec=NULL; @@ -1020,6 +998,11 @@ static TRP_RC trps_handle_update(TRPS_INSTANCE *trps, TRP_UPD *upd) } for (rec=trp_upd_get_inforec(upd); rec!=NULL; rec=trp_inforec_get_next(rec)) { + if (!trps_filter_inbound_inforec(trps, upd, rec)) { + tr_debug("trps_handle_update: inforec rejected by filter."); + continue; /* just go on to the next record */ + } + switch (trp_inforec_get_type(rec)) { case TRP_INFOREC_TYPE_ROUTE: tr_debug("trps_handle_update: handling route inforec."); @@ -1030,6 +1013,7 @@ static TRP_RC trps_handle_update(TRPS_INSTANCE *trps, TRP_UPD *upd) tr_debug("trps_handle_update: handling community inforec."); if (TRP_SUCCESS!=trps_handle_inforec_comm(trps, upd, rec)) tr_notice("trps_handle_update: error handling community inforec."); + break; default: tr_notice("trps_handle_update: unsupported inforec in TRP update."); @@ -1154,7 +1138,7 @@ TRP_RC trps_sweep_routes(TRPS_INSTANCE *trps) size_t ii=0; /* use a single time for the entire sweep */ - if (0!=clock_gettime(CLOCK_REALTIME, &sweep_time)) { + if (0!=clock_gettime(TRP_CLOCK, &sweep_time)) { tr_err("trps_sweep_routes: could not read realtime clock."); sweep_time.tv_sec=0; sweep_time.tv_nsec=0; @@ -1218,7 +1202,7 @@ TRP_RC trps_sweep_ctable(TRPS_INSTANCE *trps) TRP_RC rc=TRP_ERROR; /* use a single time for the entire sweep */ - if (0!=clock_gettime(CLOCK_REALTIME, &sweep_time)) { + if (0!=clock_gettime(TRP_CLOCK, &sweep_time)) { tr_err("trps_sweep_ctable: could not read realtime clock."); sweep_time.tv_sec=0; sweep_time.tv_nsec=0; @@ -1252,7 +1236,8 @@ TRP_RC trps_sweep_ctable(TRPS_INSTANCE *trps) /* This is the first expiration. Note this and reset the expiry time. */ tr_comm_memb_expire(memb); trps_compute_expiry(trps, tr_comm_memb_get_interval(memb), tr_comm_memb_get_expiry(memb)); - tr_debug("trps_sweep_ctable: community membership expired, resetting expiry to %s (%.*s in %.*s, origin %.*s).", + tr_debug("trps_sweep_ctable: community membership expired at %s, resetting expiry to %s (%.*s in %.*s, origin %.*s).", + timespec_to_str(&sweep_time), timespec_to_str(tr_comm_memb_get_expiry(memb)), tr_comm_memb_get_realm_id(memb)->len, tr_comm_memb_get_realm_id(memb)->buf, tr_comm_get_id(tr_comm_memb_get_comm(memb))->len, tr_comm_get_id(tr_comm_memb_get_comm(memb))->buf, @@ -1607,6 +1592,58 @@ cleanup: return rc; } +/** + * Filter the inforecs in a single update + * + * @param filt The filter to apply + * @param upd The update to filter + */ +static void trps_filter_one_outbound_update(TR_FILTER *filt, TRP_UPD *upd) +{ + TRP_INFOREC *this=NULL, *next=NULL; + TR_FILTER_ACTION action=TR_FILTER_ACTION_REJECT; + TR_FILTER_TARGET *target=NULL; + + for(this=trp_upd_get_inforec(upd); this!=NULL; this=next) { + next=this->next; + target= tr_filter_target_trp_inforec(NULL, upd, this); + if (target==NULL) { + /* TODO: signal that filtering failed. Until then, just filter everything and give an error message. */ + tr_crit("trps_filter_one_outbound_update: Unable to allocate filter target, cannot apply filter!"); + } + if ((target==NULL) + || (TR_FILTER_NO_MATCH==tr_filter_apply(target, filt, NULL, &action)) + || (action!=TR_FILTER_ACTION_ACCEPT)) { + /* Either no filter matched or one matched and rejected this record. + * Also filter out record if we were unable to allocate a target. */ + trp_upd_remove_inforec(upd, this); /* "this" is now invalid */ + } + if (target!=NULL) + tr_filter_target_free(target); + } +} + +/** + * May shuffle the update list. + * + * @param filters The filter set for the relevant TRP peer + * @param updates GPtrArray of updates to filter + */ +static void trps_filter_outbound_updates(TR_FILTER_SET *filters, GPtrArray *updates) +{ + TRP_UPD *upd=NULL; + guint ii=0; + + /* Walk backward through the array so we can remove elements. Careful about loop + * termination - remember that ii is unsigned. */ + for (ii=updates->len; ii>0; ii--) { + upd=g_ptr_array_index(updates, ii-1); + trps_filter_one_outbound_update(tr_filter_set_get(filters, TR_FILTER_TYPE_TRP_OUTBOUND), upd); + /* see if we removed all the records from this update */ + if (trp_upd_num_inforecs(upd)==0) + g_ptr_array_remove_index_fast(updates, ii-1); /* does not preserve order at index ii or higher */ + } +} /* helper for trps_update_one_peer. Frees the TRP_UPD pointed to by a GPtrArray element */ static void trps_trp_upd_destroy(gpointer data) @@ -1618,8 +1655,8 @@ static void trps_trp_upd_destroy(gpointer data) static TRP_RC trps_update_one_peer(TRPS_INSTANCE *trps, TRP_PEER *peer, TRP_UPDATE_TYPE update_type, - TR_NAME *comm, - TR_NAME *realm) + TR_NAME *realm, + TR_NAME *comm) { TALLOC_CTX *tmp_ctx=talloc_new(NULL); TR_MSG msg; /* not a pointer! */ @@ -1700,26 +1737,33 @@ static TRP_RC trps_update_one_peer(TRPS_INSTANCE *trps, if (updates->len<=0) tr_debug("trps_update_one_peer: no updates for %.*s", peer_label->len, peer_label->buf); else { - tr_debug("trps_update_one_peer: sending %d update messages.", updates->len); - for (ii=0; iilen; ii++) { - upd=(TRP_UPD *)g_ptr_array_index(updates, ii); - /* now encode the update message */ - tr_msg_set_trp_upd(&msg, upd); - encoded=tr_msg_encode(&msg); - if (encoded==NULL) { - tr_err("trps_update_one_peer: error encoding update."); - rc=TRP_ERROR; - goto cleanup; - } + /* Apply outbound TRP filters for this peer */ + trps_filter_outbound_updates(peer->filters, updates); - tr_debug("trps_update_one_peer: adding message to queue."); - if (trps_send_msg(trps, peer, encoded) != TRP_SUCCESS) - tr_err("trps_update_one_peer: error queueing update."); - else - tr_debug("trps_update_one_peer: update queued successfully."); + if (updates->len<=0) + tr_debug("trps_update_one_peer: no updates for %.*s after filtering.", peer_label->len, peer_label->buf); + else { + tr_debug("trps_update_one_peer: sending %d update messages.", updates->len); + for (ii=0; iilen; ii++) { + upd = (TRP_UPD *) g_ptr_array_index(updates, ii); + /* now encode the update message */ + tr_msg_set_trp_upd(&msg, upd); + encoded = tr_msg_encode(NULL, &msg); + if (encoded == NULL) { + tr_err("trps_update_one_peer: error encoding update."); + rc = TRP_ERROR; + goto cleanup; + } + + tr_debug("trps_update_one_peer: adding message to queue."); + if (trps_send_msg(trps, peer, encoded) != TRP_SUCCESS) + tr_err("trps_update_one_peer: error queueing update."); + else + tr_debug("trps_update_one_peer: update queued successfully."); - tr_msg_free_encoded(encoded); - encoded=NULL; + tr_msg_free_encoded(encoded); + encoded = NULL; + } } } @@ -1839,8 +1883,8 @@ static TRP_RC trps_handle_request(TRPS_INSTANCE *trps, TRP_REQ *req) return trps_update_one_peer(trps, trps_get_peer_by_gssname(trps, trp_req_get_peer(req)), TRP_UPDATE_REQUESTED, - comm, - realm); + realm, + comm); } @@ -1889,7 +1933,7 @@ TRP_RC trps_wildcard_route_req(TRPS_INSTANCE *trps, TR_NAME *peer_servicename) } tr_msg_set_trp_req(&msg, req); - encoded=tr_msg_encode(&msg); + encoded= tr_msg_encode(NULL, &msg); if (encoded==NULL) { tr_err("trps_wildcard_route_req: error encoding wildcard TRP request."); rc=TRP_ERROR;