/* Unix SMB/CIFS implementation. Samba utility functions Copyright (C) Andrew Tridgell 1992-1998 Copyright (C) Tim Potter 2000-2001 Copyright (C) Jeremy Allison 1992-2007 This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program. If not, see . */ #include "includes.h" /**************************************************************************** Get a port number in host byte order from a sockaddr_storage. ****************************************************************************/ uint16_t get_sockaddr_port(const struct sockaddr_storage *pss) { uint16_t port = 0; if (pss->ss_family != AF_INET) { #if defined(HAVE_IPV6) /* IPv6 */ const struct sockaddr_in6 *sa6 = (const struct sockaddr_in6 *)pss; port = ntohs(sa6->sin6_port); #endif } else { const struct sockaddr_in *sa = (const struct sockaddr_in *)pss; port = ntohs(sa->sin_port); } return port; } /**************************************************************************** Print out an IPv4 or IPv6 address from a struct sockaddr_storage. ****************************************************************************/ static char *print_sockaddr_len(char *dest, size_t destlen, const struct sockaddr *psa, socklen_t psalen) { if (destlen > 0) { dest[0] = '\0'; } (void)sys_getnameinfo(psa, psalen, dest, destlen, NULL, 0, NI_NUMERICHOST); return dest; } /**************************************************************************** Print out an IPv4 or IPv6 address from a struct sockaddr_storage. ****************************************************************************/ char *print_sockaddr(char *dest, size_t destlen, const struct sockaddr_storage *psa) { return print_sockaddr_len(dest, destlen, (struct sockaddr *)psa, sizeof(struct sockaddr_storage)); } /**************************************************************************** Print out a canonical IPv4 or IPv6 address from a struct sockaddr_storage. ****************************************************************************/ char *print_canonical_sockaddr(TALLOC_CTX *ctx, const struct sockaddr_storage *pss) { char addr[INET6_ADDRSTRLEN]; char *dest = NULL; int ret; /* Linux getnameinfo() man pages says port is unitialized if service name is NULL. */ ret = sys_getnameinfo((const struct sockaddr *)pss, sizeof(struct sockaddr_storage), addr, sizeof(addr), NULL, 0, NI_NUMERICHOST); if (ret != 0) { return NULL; } if (pss->ss_family != AF_INET) { #if defined(HAVE_IPV6) dest = talloc_asprintf(ctx, "[%s]", addr); #else return NULL; #endif } else { dest = talloc_asprintf(ctx, "%s", addr); } return dest; } /**************************************************************************** Return the string of an IP address (IPv4 or IPv6). ****************************************************************************/ static const char *get_socket_addr(int fd, char *addr_buf, size_t addr_len) { struct sockaddr_storage sa; socklen_t length = sizeof(sa); /* Ok, returning a hard coded IPv4 address * is bogus, but it's just as bogus as a * zero IPv6 address. No good choice here. */ strlcpy(addr_buf, "0.0.0.0", addr_len); if (fd == -1) { return addr_buf; } if (getsockname(fd, (struct sockaddr *)&sa, &length) < 0) { DEBUG(0,("getsockname failed. Error was %s\n", strerror(errno) )); return addr_buf; } return print_sockaddr_len(addr_buf, addr_len, (struct sockaddr *)&sa, length); } /**************************************************************************** Return the port number we've bound to on a socket. ****************************************************************************/ int get_socket_port(int fd) { struct sockaddr_storage sa; socklen_t length = sizeof(sa); if (fd == -1) { return -1; } if (getsockname(fd, (struct sockaddr *)&sa, &length) < 0) { DEBUG(0,("getpeername failed. Error was %s\n", strerror(errno) )); return -1; } #if defined(HAVE_IPV6) if (sa.ss_family == AF_INET6) { return ntohs(((struct sockaddr_in6 *)&sa)->sin6_port); } #endif if (sa.ss_family == AF_INET) { return ntohs(((struct sockaddr_in *)&sa)->sin_port); } return -1; } const char *client_name(int fd) { return get_peer_name(fd,false); } const char *client_addr(int fd, char *addr, size_t addrlen) { return get_peer_addr(fd,addr,addrlen); } const char *client_socket_addr(int fd, char *addr, size_t addr_len) { return get_socket_addr(fd, addr, addr_len); } #if 0 /* Not currently used. JRA. */ int client_socket_port(int fd) { return get_socket_port(fd); } #endif /**************************************************************************** Accessor functions to make thread-safe code easier later... ****************************************************************************/ void set_smb_read_error(enum smb_read_errors *pre, enum smb_read_errors newerr) { if (pre) { *pre = newerr; } } void cond_set_smb_read_error(enum smb_read_errors *pre, enum smb_read_errors newerr) { if (pre && *pre == SMB_READ_OK) { *pre = newerr; } } /**************************************************************************** Determine if a file descriptor is in fact a socket. ****************************************************************************/ bool is_a_socket(int fd) { int v; socklen_t l; l = sizeof(int); return(getsockopt(fd, SOL_SOCKET, SO_TYPE, (char *)&v, &l) == 0); } enum SOCK_OPT_TYPES {OPT_BOOL,OPT_INT,OPT_ON}; typedef struct smb_socket_option { const char *name; int level; int option; int value; int opttype; } smb_socket_option; static const smb_socket_option socket_options[] = { {"SO_KEEPALIVE", SOL_SOCKET, SO_KEEPALIVE, 0, OPT_BOOL}, {"SO_REUSEADDR", SOL_SOCKET, SO_REUSEADDR, 0, OPT_BOOL}, {"SO_BROADCAST", SOL_SOCKET, SO_BROADCAST, 0, OPT_BOOL}, #ifdef TCP_NODELAY {"TCP_NODELAY", IPPROTO_TCP, TCP_NODELAY, 0, OPT_BOOL}, #endif #ifdef TCP_KEEPCNT {"TCP_KEEPCNT", IPPROTO_TCP, TCP_KEEPCNT, 0, OPT_INT}, #endif #ifdef TCP_KEEPIDLE {"TCP_KEEPIDLE", IPPROTO_TCP, TCP_KEEPIDLE, 0, OPT_INT}, #endif #ifdef TCP_KEEPINTVL {"TCP_KEEPINTVL", IPPROTO_TCP, TCP_KEEPINTVL, 0, OPT_INT}, #endif #ifdef IPTOS_LOWDELAY {"IPTOS_LOWDELAY", IPPROTO_IP, IP_TOS, IPTOS_LOWDELAY, OPT_ON}, #endif #ifdef IPTOS_THROUGHPUT {"IPTOS_THROUGHPUT", IPPROTO_IP, IP_TOS, IPTOS_THROUGHPUT, OPT_ON}, #endif #ifdef SO_REUSEPORT {"SO_REUSEPORT", SOL_SOCKET, SO_REUSEPORT, 0, OPT_BOOL}, #endif #ifdef SO_SNDBUF {"SO_SNDBUF", SOL_SOCKET, SO_SNDBUF, 0, OPT_INT}, #endif #ifdef SO_RCVBUF {"SO_RCVBUF", SOL_SOCKET, SO_RCVBUF, 0, OPT_INT}, #endif #ifdef SO_SNDLOWAT {"SO_SNDLOWAT", SOL_SOCKET, SO_SNDLOWAT, 0, OPT_INT}, #endif #ifdef SO_RCVLOWAT {"SO_RCVLOWAT", SOL_SOCKET, SO_RCVLOWAT, 0, OPT_INT}, #endif #ifdef SO_SNDTIMEO {"SO_SNDTIMEO", SOL_SOCKET, SO_SNDTIMEO, 0, OPT_INT}, #endif #ifdef SO_RCVTIMEO {"SO_RCVTIMEO", SOL_SOCKET, SO_RCVTIMEO, 0, OPT_INT}, #endif #ifdef TCP_FASTACK {"TCP_FASTACK", IPPROTO_TCP, TCP_FASTACK, 0, OPT_INT}, #endif #ifdef TCP_QUICKACK {"TCP_QUICKACK", IPPROTO_TCP, TCP_QUICKACK, 0, OPT_BOOL}, #endif {NULL,0,0,0,0}}; /**************************************************************************** Print socket options. ****************************************************************************/ static void print_socket_options(int s) { int value; socklen_t vlen = 4; const smb_socket_option *p = &socket_options[0]; /* wrapped in if statement to prevent streams * leak in SCO Openserver 5.0 */ /* reported on samba-technical --jerry */ if ( DEBUGLEVEL >= 5 ) { DEBUG(5,("Socket options:\n")); for (; p->name != NULL; p++) { if (getsockopt(s, p->level, p->option, (void *)&value, &vlen) == -1) { DEBUGADD(5,("\tCould not test socket option %s.\n", p->name)); } else { DEBUGADD(5,("\t%s = %d\n", p->name,value)); } } } } /**************************************************************************** Set user socket options. ****************************************************************************/ void set_socket_options(int fd, const char *options) { TALLOC_CTX *ctx = talloc_stackframe(); char *tok; while (next_token_talloc(ctx, &options, &tok," \t,")) { int ret=0,i; int value = 1; char *p; bool got_value = false; if ((p = strchr_m(tok,'='))) { *p = 0; value = atoi(p+1); got_value = true; } for (i=0;socket_options[i].name;i++) if (strequal(socket_options[i].name,tok)) break; if (!socket_options[i].name) { DEBUG(0,("Unknown socket option %s\n",tok)); continue; } switch (socket_options[i].opttype) { case OPT_BOOL: case OPT_INT: ret = setsockopt(fd,socket_options[i].level, socket_options[i].option, (char *)&value,sizeof(int)); break; case OPT_ON: if (got_value) DEBUG(0,("syntax error - %s " "does not take a value\n",tok)); { int on = socket_options[i].value; ret = setsockopt(fd,socket_options[i].level, socket_options[i].option, (char *)&on,sizeof(int)); } break; } if (ret != 0) { /* be aware that some systems like Solaris return * EINVAL to a setsockopt() call when the client * sent a RST previously - no need to worry */ DEBUG(2,("Failed to set socket option %s (Error %s)\n", tok, strerror(errno) )); } } TALLOC_FREE(ctx); print_socket_options(fd); } /**************************************************************************** Read from a socket. ****************************************************************************/ ssize_t read_udp_v4_socket(int fd, char *buf, size_t len, struct sockaddr_storage *psa) { ssize_t ret; socklen_t socklen = sizeof(*psa); struct sockaddr_in *si = (struct sockaddr_in *)psa; memset((char *)psa,'\0',socklen); ret = (ssize_t)sys_recvfrom(fd,buf,len,0, (struct sockaddr *)psa,&socklen); if (ret <= 0) { /* Don't print a low debug error for a non-blocking socket. */ if (errno == EAGAIN) { DEBUG(10,("read_udp_v4_socket: returned EAGAIN\n")); } else { DEBUG(2,("read_udp_v4_socket: failed. errno=%s\n", strerror(errno))); } return 0; } if (psa->ss_family != AF_INET) { DEBUG(2,("read_udp_v4_socket: invalid address family %d " "(not IPv4)\n", (int)psa->ss_family)); return 0; } DEBUG(10,("read_udp_v4_socket: ip %s port %d read: %lu\n", inet_ntoa(si->sin_addr), si->sin_port, (unsigned long)ret)); return ret; } /**************************************************************************** Read data from a file descriptor with a timout in msec. mincount = if timeout, minimum to read before returning maxcount = number to be read. time_out = timeout in milliseconds NB. This can be called with a non-socket fd, don't change sys_read() to sys_recv() or other socket call. ****************************************************************************/ NTSTATUS read_fd_with_timeout(int fd, char *buf, size_t mincnt, size_t maxcnt, unsigned int time_out, size_t *size_ret) { fd_set fds; int selrtn; ssize_t readret; size_t nread = 0; struct timeval timeout; char addr[INET6_ADDRSTRLEN]; int save_errno; /* just checking .... */ if (maxcnt <= 0) return NT_STATUS_OK; /* Blocking read */ if (time_out == 0) { if (mincnt == 0) { mincnt = maxcnt; } while (nread < mincnt) { readret = sys_read(fd, buf + nread, maxcnt - nread); if (readret == 0) { DEBUG(5,("read_fd_with_timeout: " "blocking read. EOF from client.\n")); return NT_STATUS_END_OF_FILE; } if (readret == -1) { save_errno = errno; if (fd == get_client_fd()) { /* Try and give an error message * saying what client failed. */ DEBUG(0,("read_fd_with_timeout: " "client %s read error = %s.\n", get_peer_addr(fd,addr,sizeof(addr)), strerror(save_errno) )); } else { DEBUG(0,("read_fd_with_timeout: " "read error = %s.\n", strerror(save_errno) )); } return map_nt_error_from_unix(save_errno); } nread += readret; } goto done; } /* Most difficult - timeout read */ /* If this is ever called on a disk file and mincnt is greater then the filesize then system performance will suffer severely as select always returns true on disk files */ /* Set initial timeout */ timeout.tv_sec = (time_t)(time_out / 1000); timeout.tv_usec = (long)(1000 * (time_out % 1000)); for (nread=0; nread < mincnt; ) { FD_ZERO(&fds); FD_SET(fd,&fds); selrtn = sys_select_intr(fd+1,&fds,NULL,NULL,&timeout); /* Check if error */ if (selrtn == -1) { save_errno = errno; /* something is wrong. Maybe the socket is dead? */ if (fd == get_client_fd()) { /* Try and give an error message saying * what client failed. */ DEBUG(0,("read_fd_with_timeout: timeout " "read for client %s. select error = %s.\n", get_peer_addr(fd,addr,sizeof(addr)), strerror(save_errno) )); } else { DEBUG(0,("read_fd_with_timeout: timeout " "read. select error = %s.\n", strerror(save_errno) )); } return map_nt_error_from_unix(save_errno); } /* Did we timeout ? */ if (selrtn == 0) { DEBUG(10,("read_fd_with_timeout: timeout read. " "select timed out.\n")); return NT_STATUS_IO_TIMEOUT; } readret = sys_read(fd, buf+nread, maxcnt-nread); if (readret == 0) { /* we got EOF on the file descriptor */ DEBUG(5,("read_fd_with_timeout: timeout read. " "EOF from client.\n")); return NT_STATUS_END_OF_FILE; } if (readret == -1) { save_errno = errno; /* the descriptor is probably dead */ if (fd == get_client_fd()) { /* Try and give an error message * saying what client failed. */ DEBUG(0,("read_fd_with_timeout: timeout " "read to client %s. read error = %s.\n", get_peer_addr(fd,addr,sizeof(addr)), strerror(save_errno) )); } else { DEBUG(0,("read_fd_with_timeout: timeout " "read. read error = %s.\n", strerror(save_errno) )); } return map_nt_error_from_unix(errno); } nread += readret; } done: /* Return the number we got */ if (size_ret) { *size_ret = nread; } return NT_STATUS_OK; } /**************************************************************************** Read data from an fd, reading exactly N bytes. NB. This can be called with a non-socket fd, don't add dependencies on socket calls. ****************************************************************************/ NTSTATUS read_data(int fd, char *buffer, size_t N) { return read_fd_with_timeout(fd, buffer, N, N, 0, NULL); } /**************************************************************************** Write all data from an iov array NB. This can be called with a non-socket fd, don't add dependencies on socket calls. ****************************************************************************/ ssize_t write_data_iov(int fd, const struct iovec *orig_iov, int iovcnt) { int i; size_t to_send; ssize_t thistime; size_t sent; struct iovec *iov_copy, *iov; to_send = 0; for (i=0; i 0) { if (thistime < iov[0].iov_len) { char *new_base = (char *)iov[0].iov_base + thistime; iov[0].iov_base = (void *)new_base; iov[0].iov_len -= thistime; break; } thistime -= iov[0].iov_len; iov += 1; iovcnt -= 1; } thistime = sys_writev(fd, iov, iovcnt); if (thistime <= 0) { break; } sent += thistime; } TALLOC_FREE(iov_copy); return sent; } /**************************************************************************** Write data to a fd. NB. This can be called with a non-socket fd, don't add dependencies on socket calls. ****************************************************************************/ ssize_t write_data(int fd, const char *buffer, size_t N) { ssize_t ret; struct iovec iov; iov.iov_base = CONST_DISCARD(void *, buffer); iov.iov_len = N; ret = write_data_iov(fd, &iov, 1); if (ret >= 0) { return ret; } if (fd == get_client_fd()) { char addr[INET6_ADDRSTRLEN]; /* * Try and give an error message saying what client failed. */ DEBUG(0, ("write_data: write failure in writing to client %s. " "Error %s\n", get_peer_addr(fd,addr,sizeof(addr)), strerror(errno))); } else { DEBUG(0,("write_data: write failure. Error = %s\n", strerror(errno) )); } return -1; } /**************************************************************************** Send a keepalive packet (rfc1002). ****************************************************************************/ bool send_keepalive(int client) { unsigned char buf[4]; buf[0] = SMBkeepalive; buf[1] = buf[2] = buf[3] = 0; return(write_data(client,(char *)buf,4) == 4); } /**************************************************************************** Read 4 bytes of a smb packet and return the smb length of the packet. Store the result in the buffer. This version of the function will return a length of zero on receiving a keepalive packet. Timeout is in milliseconds. ****************************************************************************/ NTSTATUS read_smb_length_return_keepalive(int fd, char *inbuf, unsigned int timeout, size_t *len) { int msg_type; NTSTATUS status; status = read_fd_with_timeout(fd, inbuf, 4, 4, timeout, NULL); if (!NT_STATUS_IS_OK(status)) { return status; } *len = smb_len(inbuf); msg_type = CVAL(inbuf,0); if (msg_type == SMBkeepalive) { DEBUG(5,("Got keepalive packet\n")); } DEBUG(10,("got smb length of %lu\n",(unsigned long)(*len))); return NT_STATUS_OK; } /**************************************************************************** Read 4 bytes of a smb packet and return the smb length of the packet. Store the result in the buffer. This version of the function will never return a session keepalive (length of zero). Timeout is in milliseconds. ****************************************************************************/ NTSTATUS read_smb_length(int fd, char *inbuf, unsigned int timeout, size_t *len) { uint8_t msgtype = SMBkeepalive; while (msgtype == SMBkeepalive) { NTSTATUS status; status = read_smb_length_return_keepalive(fd, inbuf, timeout, len); if (!NT_STATUS_IS_OK(status)) { return status; } msgtype = CVAL(inbuf, 0); } DEBUG(10,("read_smb_length: got smb length of %lu\n", (unsigned long)len)); return NT_STATUS_OK; } /**************************************************************************** Read an smb from a fd. The timeout is in milliseconds. This function will return on receipt of a session keepalive packet. maxlen is the max number of bytes to return, not including the 4 byte length. If zero it means buflen limit. Doesn't check the MAC on signed packets. ****************************************************************************/ NTSTATUS receive_smb_raw(int fd, char *buffer, size_t buflen, unsigned int timeout, size_t maxlen, size_t *p_len) { size_t len; NTSTATUS status; status = read_smb_length_return_keepalive(fd,buffer,timeout,&len); if (!NT_STATUS_IS_OK(status)) { DEBUG(10, ("receive_smb_raw: %s!\n", nt_errstr(status))); return status; } if (len > buflen) { DEBUG(0,("Invalid packet length! (%lu bytes).\n", (unsigned long)len)); return NT_STATUS_INVALID_PARAMETER; } if(len > 0) { if (maxlen) { len = MIN(len,maxlen); } status = read_fd_with_timeout( fd, buffer+4, len, len, timeout, &len); if (!NT_STATUS_IS_OK(status)) { return status; } /* not all of samba3 properly checks for packet-termination * of strings. This ensures that we don't run off into * empty space. */ SSVAL(buffer+4,len, 0); } *p_len = len; return NT_STATUS_OK; } /**************************************************************************** Open a socket of the specified type, port, and address for incoming data. ****************************************************************************/ int open_socket_in(int type, uint16_t port, int dlevel, const struct sockaddr_storage *psock, bool rebind) { struct sockaddr_storage sock; int res; socklen_t slen = sizeof(struct sockaddr_in); sock = *psock; #if defined(HAVE_IPV6) if (sock.ss_family == AF_INET6) { ((struct sockaddr_in6 *)&sock)->sin6_port = htons(port); slen = sizeof(struct sockaddr_in6); } #endif if (sock.ss_family == AF_INET) { ((struct sockaddr_in *)&sock)->sin_port = htons(port); } res = socket(sock.ss_family, type, 0 ); if( res == -1 ) { if( DEBUGLVL(0) ) { dbgtext( "open_socket_in(): socket() call failed: " ); dbgtext( "%s\n", strerror( errno ) ); } return -1; } /* This block sets/clears the SO_REUSEADDR and possibly SO_REUSEPORT. */ { int val = rebind ? 1 : 0; if( setsockopt(res,SOL_SOCKET,SO_REUSEADDR, (char *)&val,sizeof(val)) == -1 ) { if( DEBUGLVL( dlevel ) ) { dbgtext( "open_socket_in(): setsockopt: " ); dbgtext( "SO_REUSEADDR = %s ", val?"true":"false" ); dbgtext( "on port %d failed ", port ); dbgtext( "with error = %s\n", strerror(errno) ); } } #ifdef SO_REUSEPORT if( setsockopt(res,SOL_SOCKET,SO_REUSEPORT, (char *)&val,sizeof(val)) == -1 ) { if( DEBUGLVL( dlevel ) ) { dbgtext( "open_socket_in(): setsockopt: "); dbgtext( "SO_REUSEPORT = %s ", val?"true":"false"); dbgtext( "on port %d failed ", port); dbgtext( "with error = %s\n", strerror(errno)); } } #endif /* SO_REUSEPORT */ } /* now we've got a socket - we need to bind it */ if (bind(res, (struct sockaddr *)&sock, slen) == -1 ) { if( DEBUGLVL(dlevel) && (port == SMB_PORT1 || port == SMB_PORT2 || port == NMB_PORT) ) { char addr[INET6_ADDRSTRLEN]; print_sockaddr(addr, sizeof(addr), &sock); dbgtext( "bind failed on port %d ", port); dbgtext( "socket_addr = %s.\n", addr); dbgtext( "Error = %s\n", strerror(errno)); } close(res); return -1; } DEBUG( 10, ( "bind succeeded on port %d\n", port ) ); return( res ); } struct open_socket_out_state { int fd; struct event_context *ev; struct sockaddr_storage ss; socklen_t salen; uint16_t port; int wait_nsec; }; static void open_socket_out_connected(struct tevent_req *subreq); static int open_socket_out_state_destructor(struct open_socket_out_state *s) { if (s->fd != -1) { close(s->fd); } return 0; } /**************************************************************************** Create an outgoing socket. timeout is in milliseconds. **************************************************************************/ struct tevent_req *open_socket_out_send(TALLOC_CTX *mem_ctx, struct event_context *ev, const struct sockaddr_storage *pss, uint16_t port, int timeout) { char addr[INET6_ADDRSTRLEN]; struct tevent_req *result, *subreq; struct open_socket_out_state *state; NTSTATUS status; result = tevent_req_create(mem_ctx, &state, struct open_socket_out_state); if (result == NULL) { return NULL; } state->ev = ev; state->ss = *pss; state->port = port; state->wait_nsec = 10000; state->salen = -1; state->fd = socket(state->ss.ss_family, SOCK_STREAM, 0); if (state->fd == -1) { status = map_nt_error_from_unix(errno); goto post_status; } talloc_set_destructor(state, open_socket_out_state_destructor); if (!tevent_req_set_endtime( result, ev, timeval_current_ofs(0, timeout*1000))) { goto fail; } #if defined(HAVE_IPV6) if (pss->ss_family == AF_INET6) { struct sockaddr_in6 *psa6; psa6 = (struct sockaddr_in6 *)&state->ss; psa6->sin6_port = htons(port); if (psa6->sin6_scope_id == 0 && IN6_IS_ADDR_LINKLOCAL(&psa6->sin6_addr)) { setup_linklocal_scope_id( (struct sockaddr *)&(state->ss)); } state->salen = sizeof(struct sockaddr_in6); } #endif if (pss->ss_family == AF_INET) { struct sockaddr_in *psa; psa = (struct sockaddr_in *)&state->ss; psa->sin_port = htons(port); state->salen = sizeof(struct sockaddr_in); } if (pss->ss_family == AF_UNIX) { state->salen = sizeof(struct sockaddr_un); } print_sockaddr(addr, sizeof(addr), &state->ss); DEBUG(3,("Connecting to %s at port %u\n", addr, (unsigned int)port)); subreq = async_connect_send(state, state->ev, state->fd, (struct sockaddr *)&state->ss, state->salen); if ((subreq == NULL) || !tevent_req_set_endtime( subreq, state->ev, timeval_current_ofs(0, state->wait_nsec))) { goto fail; } tevent_req_set_callback(subreq, open_socket_out_connected, result); return result; post_status: tevent_req_nterror(result, status); return tevent_req_post(result, ev); fail: TALLOC_FREE(result); return NULL; } static void open_socket_out_connected(struct tevent_req *subreq) { struct tevent_req *req = tevent_req_callback_data(subreq, struct tevent_req); struct open_socket_out_state *state = tevent_req_data(req, struct open_socket_out_state); int ret; int sys_errno; ret = async_connect_recv(subreq, &sys_errno); TALLOC_FREE(subreq); if (ret == 0) { tevent_req_done(req); return; } if ( #ifdef ETIMEDOUT (sys_errno == ETIMEDOUT) || #endif (sys_errno == EINPROGRESS) || (sys_errno == EALREADY) || (sys_errno == EAGAIN)) { /* * retry */ if (state->wait_nsec < 250000) { state->wait_nsec *= 1.5; } subreq = async_connect_send(state, state->ev, state->fd, (struct sockaddr *)&state->ss, state->salen); if (tevent_req_nomem(subreq, req)) { return; } if (!tevent_req_set_endtime( subreq, state->ev, timeval_current_ofs(0, state->wait_nsec))) { tevent_req_nterror(req, NT_STATUS_NO_MEMORY); return; } tevent_req_set_callback(subreq, open_socket_out_connected, req); return; } #ifdef EISCONN if (sys_errno == EISCONN) { tevent_req_done(req); return; } #endif /* real error */ tevent_req_nterror(req, map_nt_error_from_unix(sys_errno)); } NTSTATUS open_socket_out_recv(struct tevent_req *req, int *pfd) { struct open_socket_out_state *state = tevent_req_data(req, struct open_socket_out_state); NTSTATUS status; if (tevent_req_is_nterror(req, &status)) { return status; } *pfd = state->fd; state->fd = -1; return NT_STATUS_OK; } NTSTATUS open_socket_out(const struct sockaddr_storage *pss, uint16_t port, int timeout, int *pfd) { TALLOC_CTX *frame = talloc_stackframe(); struct event_context *ev; struct tevent_req *req; NTSTATUS status = NT_STATUS_NO_MEMORY; ev = event_context_init(frame); if (ev == NULL) { goto fail; } req = open_socket_out_send(frame, ev, pss, port, timeout); if (req == NULL) { goto fail; } if (!tevent_req_poll(req, ev)) { status = NT_STATUS_INTERNAL_ERROR; goto fail; } status = open_socket_out_recv(req, pfd); fail: TALLOC_FREE(frame); return status; } struct open_socket_out_defer_state { struct event_context *ev; struct sockaddr_storage ss; uint16_t port; int timeout; int fd; }; static void open_socket_out_defer_waited(struct tevent_req *subreq); static void open_socket_out_defer_connected(struct tevent_req *subreq); struct tevent_req *open_socket_out_defer_send(TALLOC_CTX *mem_ctx, struct event_context *ev, struct timeval wait_time, const struct sockaddr_storage *pss, uint16_t port, int timeout) { struct tevent_req *req, *subreq; struct open_socket_out_defer_state *state; req = tevent_req_create(mem_ctx, &state, struct open_socket_out_defer_state); if (req == NULL) { return NULL; } state->ev = ev; state->ss = *pss; state->port = port; state->timeout = timeout; subreq = tevent_wakeup_send( state, ev, timeval_current_ofs(wait_time.tv_sec, wait_time.tv_usec)); if (subreq == NULL) { goto fail; } tevent_req_set_callback(subreq, open_socket_out_defer_waited, req); return req; fail: TALLOC_FREE(req); return NULL; } static void open_socket_out_defer_waited(struct tevent_req *subreq) { struct tevent_req *req = tevent_req_callback_data( subreq, struct tevent_req); struct open_socket_out_defer_state *state = tevent_req_data( req, struct open_socket_out_defer_state); bool ret; ret = tevent_wakeup_recv(subreq); TALLOC_FREE(subreq); if (!ret) { tevent_req_nterror(req, NT_STATUS_INTERNAL_ERROR); return; } subreq = open_socket_out_send(state, state->ev, &state->ss, state->port, state->timeout); if (tevent_req_nomem(subreq, req)) { return; } tevent_req_set_callback(subreq, open_socket_out_defer_connected, req); } static void open_socket_out_defer_connected(struct tevent_req *subreq) { struct tevent_req *req = tevent_req_callback_data( subreq, struct tevent_req); struct open_socket_out_defer_state *state = tevent_req_data( req, struct open_socket_out_defer_state); NTSTATUS status; status = open_socket_out_recv(subreq, &state->fd); TALLOC_FREE(subreq); if (!NT_STATUS_IS_OK(status)) { tevent_req_nterror(req, status); return; } tevent_req_done(req); } NTSTATUS open_socket_out_defer_recv(struct tevent_req *req, int *pfd) { struct open_socket_out_defer_state *state = tevent_req_data( req, struct open_socket_out_defer_state); NTSTATUS status; if (tevent_req_is_nterror(req, &status)) { return status; } *pfd = state->fd; state->fd = -1; return NT_STATUS_OK; } /******************************************************************* Create an outgoing TCP socket to the first addr that connects. This is for simultaneous connection attempts to port 445 and 139 of a host or for simultatneous connection attempts to multiple DCs at once. We return a socket fd of the first successful connection. @param[in] addrs list of Internet addresses and ports to connect to @param[in] num_addrs number of address/port pairs in the addrs list @param[in] timeout time after which we stop waiting for a socket connection to succeed, given in milliseconds @param[out] fd_index the entry in addrs which we successfully connected to @param[out] fd fd of the open and connected socket @return true on a successful connection, false if all connection attempts failed or we timed out *******************************************************************/ bool open_any_socket_out(struct sockaddr_storage *addrs, int num_addrs, int timeout, int *fd_index, int *fd) { int i, resulting_index, res; int *sockets; bool good_connect; fd_set r_fds, wr_fds; struct timeval tv; int maxfd; int connect_loop = 10000; /* 10 milliseconds */ timeout *= 1000; /* convert to microseconds */ sockets = SMB_MALLOC_ARRAY(int, num_addrs); if (sockets == NULL) return false; resulting_index = -1; for (i=0; imaxfd) maxfd = sockets[i]; } tv.tv_sec = 0; tv.tv_usec = connect_loop; res = sys_select_intr(maxfd+1, &r_fds, &wr_fds, NULL, &tv); if (res < 0) goto done; if (res == 0) goto next_round; for (i=0; i timeout) connect_loop = timeout; goto connect_again; done: for (i=0; i= 0) close(sockets[i]); } if (resulting_index >= 0) { *fd_index = resulting_index; *fd = sockets[*fd_index]; set_blocking(*fd, true); } free(sockets); return (resulting_index >= 0); } /**************************************************************************** Open a connected UDP socket to host on port **************************************************************************/ int open_udp_socket(const char *host, int port) { struct sockaddr_storage ss; int res; if (!interpret_string_addr(&ss, host, 0)) { DEBUG(10,("open_udp_socket: can't resolve name %s\n", host)); return -1; } res = socket(ss.ss_family, SOCK_DGRAM, 0); if (res == -1) { return -1; } #if defined(HAVE_IPV6) if (ss.ss_family == AF_INET6) { struct sockaddr_in6 *psa6; psa6 = (struct sockaddr_in6 *)&ss; psa6->sin6_port = htons(port); if (psa6->sin6_scope_id == 0 && IN6_IS_ADDR_LINKLOCAL(&psa6->sin6_addr)) { setup_linklocal_scope_id( (struct sockaddr *)&ss); } } #endif if (ss.ss_family == AF_INET) { struct sockaddr_in *psa; psa = (struct sockaddr_in *)&ss; psa->sin_port = htons(port); } if (sys_connect(res,(struct sockaddr *)&ss)) { close(res); return -1; } return res; } /******************************************************************* Return the IP addr of the remote end of a socket as a string. Optionally return the struct sockaddr_storage. ******************************************************************/ static const char *get_peer_addr_internal(int fd, char *addr_buf, size_t addr_buf_len, struct sockaddr *pss, socklen_t *plength) { struct sockaddr_storage ss; socklen_t length = sizeof(ss); strlcpy(addr_buf,"0.0.0.0",addr_buf_len); if (fd == -1) { return addr_buf; } if (pss == NULL) { pss = (struct sockaddr *)&ss; plength = &length; } if (getpeername(fd, (struct sockaddr *)pss, plength) < 0) { DEBUG(0,("getpeername failed. Error was %s\n", strerror(errno) )); return addr_buf; } print_sockaddr_len(addr_buf, addr_buf_len, pss, *plength); return addr_buf; } /******************************************************************* Matchname - determine if host name matches IP address. Used to confirm a hostname lookup to prevent spoof attacks. ******************************************************************/ static bool matchname(const char *remotehost, const struct sockaddr *pss, socklen_t len) { struct addrinfo *res = NULL; struct addrinfo *ailist = NULL; char addr_buf[INET6_ADDRSTRLEN]; bool ret = interpret_string_addr_internal(&ailist, remotehost, AI_ADDRCONFIG|AI_CANONNAME); if (!ret || ailist == NULL) { DEBUG(3,("matchname: getaddrinfo failed for " "name %s [%s]\n", remotehost, gai_strerror(ret) )); return false; } /* * Make sure that getaddrinfo() returns the "correct" host name. */ if (ailist->ai_canonname == NULL || (!strequal(remotehost, ailist->ai_canonname) && !strequal(remotehost, "localhost"))) { DEBUG(0,("matchname: host name/name mismatch: %s != %s\n", remotehost, ailist->ai_canonname ? ailist->ai_canonname : "(NULL)")); freeaddrinfo(ailist); return false; } /* Look up the host address in the address list we just got. */ for (res = ailist; res; res = res->ai_next) { if (!res->ai_addr) { continue; } if (sockaddr_equal((const struct sockaddr *)res->ai_addr, (struct sockaddr *)pss)) { freeaddrinfo(ailist); return true; } } /* * The host name does not map to the original host address. Perhaps * someone has compromised a name server. More likely someone botched * it, but that could be dangerous, too. */ DEBUG(0,("matchname: host name/address mismatch: %s != %s\n", print_sockaddr_len(addr_buf, sizeof(addr_buf), pss, len), ailist->ai_canonname ? ailist->ai_canonname : "(NULL)")); if (ailist) { freeaddrinfo(ailist); } return false; } /******************************************************************* Deal with the singleton cache. ******************************************************************/ struct name_addr_pair { struct sockaddr_storage ss; const char *name; }; /******************************************************************* Lookup a name/addr pair. Returns memory allocated from memcache. ******************************************************************/ static bool lookup_nc(struct name_addr_pair *nc) { DATA_BLOB tmp; ZERO_STRUCTP(nc); if (!memcache_lookup( NULL, SINGLETON_CACHE, data_blob_string_const_null("get_peer_name"), &tmp)) { return false; } memcpy(&nc->ss, tmp.data, sizeof(nc->ss)); nc->name = (const char *)tmp.data + sizeof(nc->ss); return true; } /******************************************************************* Save a name/addr pair. ******************************************************************/ static void store_nc(const struct name_addr_pair *nc) { DATA_BLOB tmp; size_t namelen = strlen(nc->name); tmp = data_blob(NULL, sizeof(nc->ss) + namelen + 1); if (!tmp.data) { return; } memcpy(tmp.data, &nc->ss, sizeof(nc->ss)); memcpy(tmp.data+sizeof(nc->ss), nc->name, namelen+1); memcache_add(NULL, SINGLETON_CACHE, data_blob_string_const_null("get_peer_name"), tmp); data_blob_free(&tmp); } /******************************************************************* Return the DNS name of the remote end of a socket. ******************************************************************/ const char *get_peer_name(int fd, bool force_lookup) { struct name_addr_pair nc; char addr_buf[INET6_ADDRSTRLEN]; struct sockaddr_storage ss; socklen_t length = sizeof(ss); const char *p; int ret; char name_buf[MAX_DNS_NAME_LENGTH]; char tmp_name[MAX_DNS_NAME_LENGTH]; /* reverse lookups can be *very* expensive, and in many situations won't work because many networks don't link dhcp with dns. To avoid the delay we avoid the lookup if possible */ if (!lp_hostname_lookups() && (force_lookup == false)) { length = sizeof(nc.ss); nc.name = get_peer_addr_internal(fd, addr_buf, sizeof(addr_buf), (struct sockaddr *)&nc.ss, &length); store_nc(&nc); lookup_nc(&nc); return nc.name ? nc.name : "UNKNOWN"; } lookup_nc(&nc); memset(&ss, '\0', sizeof(ss)); p = get_peer_addr_internal(fd, addr_buf, sizeof(addr_buf), (struct sockaddr *)&ss, &length); /* it might be the same as the last one - save some DNS work */ if (sockaddr_equal((struct sockaddr *)&ss, (struct sockaddr *)&nc.ss)) { return nc.name ? nc.name : "UNKNOWN"; } /* Not the same. We need to lookup. */ if (fd == -1) { return "UNKNOWN"; } /* Look up the remote host name. */ ret = sys_getnameinfo((struct sockaddr *)&ss, length, name_buf, sizeof(name_buf), NULL, 0, 0); if (ret) { DEBUG(1,("get_peer_name: getnameinfo failed " "for %s with error %s\n", p, gai_strerror(ret))); strlcpy(name_buf, p, sizeof(name_buf)); } else { if (!matchname(name_buf, (struct sockaddr *)&ss, length)) { DEBUG(0,("Matchname failed on %s %s\n",name_buf,p)); strlcpy(name_buf,"UNKNOWN",sizeof(name_buf)); } } /* can't pass the same source and dest strings in when you use --enable-developer or the clobber_region() call will get you */ strlcpy(tmp_name, name_buf, sizeof(tmp_name)); alpha_strcpy(name_buf, tmp_name, "_-.", sizeof(name_buf)); if (strstr(name_buf,"..")) { strlcpy(name_buf, "UNKNOWN", sizeof(name_buf)); } nc.name = name_buf; nc.ss = ss; store_nc(&nc); lookup_nc(&nc); return nc.name ? nc.name : "UNKNOWN"; } /******************************************************************* Return the IP addr of the remote end of a socket as a string. ******************************************************************/ const char *get_peer_addr(int fd, char *addr, size_t addr_len) { return get_peer_addr_internal(fd, addr, addr_len, NULL, NULL); } /******************************************************************* Create protected unix domain socket. Some unixes cannot set permissions on a ux-dom-sock, so we have to make sure that the directory contains the protection permissions instead. ******************************************************************/ int create_pipe_sock(const char *socket_dir, const char *socket_name, mode_t dir_perms) { #ifdef HAVE_UNIXSOCKET struct sockaddr_un sunaddr; struct stat st; int sock; mode_t old_umask; char *path = NULL; old_umask = umask(0); /* Create the socket directory or reuse the existing one */ if (lstat(socket_dir, &st) == -1) { if (errno == ENOENT) { /* Create directory */ if (mkdir(socket_dir, dir_perms) == -1) { DEBUG(0, ("error creating socket directory " "%s: %s\n", socket_dir, strerror(errno))); goto out_umask; } } else { DEBUG(0, ("lstat failed on socket directory %s: %s\n", socket_dir, strerror(errno))); goto out_umask; } } else { /* Check ownership and permission on existing directory */ if (!S_ISDIR(st.st_mode)) { DEBUG(0, ("socket directory %s isn't a directory\n", socket_dir)); goto out_umask; } if ((st.st_uid != sec_initial_uid()) || ((st.st_mode & 0777) != dir_perms)) { DEBUG(0, ("invalid permissions on socket directory " "%s\n", socket_dir)); goto out_umask; } } /* Create the socket file */ sock = socket(AF_UNIX, SOCK_STREAM, 0); if (sock == -1) { DEBUG(0, ("create_pipe_sock: socket error %s\n", strerror(errno) )); goto out_close; } if (asprintf(&path, "%s/%s", socket_dir, socket_name) == -1) { goto out_close; } unlink(path); memset(&sunaddr, 0, sizeof(sunaddr)); sunaddr.sun_family = AF_UNIX; strlcpy(sunaddr.sun_path, path, sizeof(sunaddr.sun_path)); if (bind(sock, (struct sockaddr *)&sunaddr, sizeof(sunaddr)) == -1) { DEBUG(0, ("bind failed on pipe socket %s: %s\n", path, strerror(errno))); goto out_close; } if (listen(sock, 5) == -1) { DEBUG(0, ("listen failed on pipe socket %s: %s\n", path, strerror(errno))); goto out_close; } SAFE_FREE(path); umask(old_umask); return sock; out_close: SAFE_FREE(path); if (sock != -1) close(sock); out_umask: umask(old_umask); return -1; #else DEBUG(0, ("create_pipe_sock: No Unix sockets on this system\n")); return -1; #endif /* HAVE_UNIXSOCKET */ } /**************************************************************************** Get my own canonical name, including domain. ****************************************************************************/ const char *get_mydnsfullname(void) { struct addrinfo *res = NULL; char my_hostname[HOST_NAME_MAX]; bool ret; DATA_BLOB tmp; if (memcache_lookup(NULL, SINGLETON_CACHE, data_blob_string_const_null("get_mydnsfullname"), &tmp)) { SMB_ASSERT(tmp.length > 0); return (const char *)tmp.data; } /* get my host name */ if (gethostname(my_hostname, sizeof(my_hostname)) == -1) { DEBUG(0,("get_mydnsfullname: gethostname failed\n")); return NULL; } /* Ensure null termination. */ my_hostname[sizeof(my_hostname)-1] = '\0'; ret = interpret_string_addr_internal(&res, my_hostname, AI_ADDRCONFIG|AI_CANONNAME); if (!ret || res == NULL) { DEBUG(3,("get_mydnsfullname: getaddrinfo failed for " "name %s [%s]\n", my_hostname, gai_strerror(ret) )); return NULL; } /* * Make sure that getaddrinfo() returns the "correct" host name. */ if (res->ai_canonname == NULL) { DEBUG(3,("get_mydnsfullname: failed to get " "canonical name for %s\n", my_hostname)); freeaddrinfo(res); return NULL; } /* This copies the data, so we must do a lookup * afterwards to find the value to return. */ memcache_add(NULL, SINGLETON_CACHE, data_blob_string_const_null("get_mydnsfullname"), data_blob_string_const_null(res->ai_canonname)); if (!memcache_lookup(NULL, SINGLETON_CACHE, data_blob_string_const_null("get_mydnsfullname"), &tmp)) { tmp = data_blob_talloc(talloc_tos(), res->ai_canonname, strlen(res->ai_canonname) + 1); } freeaddrinfo(res); return (const char *)tmp.data; } /************************************************************ Is this my name ? ************************************************************/ bool is_myname_or_ipaddr(const char *s) { TALLOC_CTX *ctx = talloc_tos(); char addr[INET6_ADDRSTRLEN]; char *name = NULL; const char *dnsname; char *servername = NULL; if (!s) { return false; } /* Santize the string from '\\name' */ name = talloc_strdup(ctx, s); if (!name) { return false; } servername = strrchr_m(name, '\\' ); if (!servername) { servername = name; } else { servername++; } /* Optimize for the common case */ if (strequal(servername, global_myname())) { return true; } /* Check for an alias */ if (is_myname(servername)) { return true; } /* Check for loopback */ if (strequal(servername, "127.0.0.1") || strequal(servername, "::1")) { return true; } if (strequal(servername, "localhost")) { return true; } /* Maybe it's my dns name */ dnsname = get_mydnsfullname(); if (dnsname && strequal(servername, dnsname)) { return true; } /* Handle possible CNAME records - convert to an IP addr. */ if (!is_ipaddress(servername)) { /* Use DNS to resolve the name, but only the first address */ struct sockaddr_storage ss; if (interpret_string_addr(&ss, servername, 0)) { print_sockaddr(addr, sizeof(addr), &ss); servername = addr; } } /* Maybe its an IP address? */ if (is_ipaddress(servername)) { struct sockaddr_storage ss; struct iface_struct *nics; int i, n; if (!interpret_string_addr(&ss, servername, AI_NUMERICHOST)) { return false; } if (ismyaddr((struct sockaddr *)&ss)) { return true; } if (is_zero_addr((struct sockaddr *)&ss) || is_loopback_addr((struct sockaddr *)&ss)) { return false; } n = get_interfaces(talloc_tos(), &nics); for (i=0; inode = node; state->service = service; state->hints = hints; subreq = fncall_send(state, ev, ctx, getaddrinfo_do, state); if (tevent_req_nomem(subreq, req)) { return tevent_req_post(req, ev); } tevent_req_set_callback(subreq, getaddrinfo_done, req); return req; } static void getaddrinfo_do(void *private_data) { struct getaddrinfo_state *state = (struct getaddrinfo_state *)private_data; state->ret = getaddrinfo(state->node, state->service, state->hints, &state->res); } static void getaddrinfo_done(struct tevent_req *subreq) { struct tevent_req *req = tevent_req_callback_data( subreq, struct tevent_req); int ret, err; ret = fncall_recv(subreq, &err); TALLOC_FREE(subreq); if (ret == -1) { tevent_req_error(req, err); return; } tevent_req_done(req); } int getaddrinfo_recv(struct tevent_req *req, struct addrinfo **res) { struct getaddrinfo_state *state = tevent_req_data( req, struct getaddrinfo_state); int err; if (tevent_req_is_unix_error(req, &err)) { switch(err) { case ENOMEM: return EAI_MEMORY; default: return EAI_FAIL; } } if (state->ret == 0) { *res = state->res; } return state->ret; } n1293' href='#n1293'>1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680
#!/usr/bin/python
'''
Created on Sep 18, 2009

@author: sgallagh
'''
import unittest
import os
from stat import *

import SSSDConfig

class SSSDConfigTestValid(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        pass

    def testServices(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                srcdir + "/etc/sssd.api.d")
        sssdconfig.import_config(srcdir + "/testconfigs/sssd-valid.conf")

        # Validate services
        services = sssdconfig.list_services()
        self.assertTrue('sssd' in services)
        self.assertTrue('nss' in services)
        self.assertTrue('pam' in services)

        #Verify service attributes
        sssd_service = sssdconfig.get_service('sssd')
        service_opts = sssd_service.list_options()


        self.assertTrue('services' in service_opts.keys())
        service_list = sssd_service.get_option('services')
        self.assertTrue('nss' in service_list)
        self.assertTrue('pam' in service_list)

        self.assertTrue('domains' in service_opts)

        self.assertTrue('reconnection_retries' in service_opts)

        del sssdconfig
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                     srcdir + "/etc/sssd.api.d")
        sssdconfig.new_config()
        sssdconfig.delete_service('sssd')
        new_sssd_service = sssdconfig.new_service('sssd');
        new_options = new_sssd_service.list_options();

        self.assertTrue('debug_level' in new_options)
        self.assertEquals(new_options['debug_level'][0], int)

        self.assertTrue('command' in new_options)
        self.assertEquals(new_options['command'][0], str)

        self.assertTrue('reconnection_retries' in new_options)
        self.assertEquals(new_options['reconnection_retries'][0], int)

        self.assertTrue('services' in new_options)
        self.assertEquals(new_options['debug_level'][0], int)

        self.assertTrue('domains' in new_options)
        self.assertEquals(new_options['domains'][0], list)
        self.assertEquals(new_options['domains'][1], str)

        self.assertTrue('sbus_timeout' in new_options)
        self.assertEquals(new_options['sbus_timeout'][0], int)

        self.assertTrue('re_expression' in new_options)
        self.assertEquals(new_options['re_expression'][0], str)

        self.assertTrue('full_name_format' in new_options)
        self.assertEquals(new_options['full_name_format'][0], str)

        del sssdconfig

    def testDomains(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                srcdir + "/etc/sssd.api.d")
        sssdconfig.import_config(srcdir + "/testconfigs/sssd-valid.conf")

        #Validate domain list
        domains = sssdconfig.list_domains()
        self.assertTrue('LOCAL' in domains)
        self.assertTrue('LDAP' in domains)
        self.assertTrue('PROXY' in domains)
        self.assertTrue('IPA' in domains)

        #Verify domain attributes
        ipa_domain = sssdconfig.get_domain('IPA')
        domain_opts = ipa_domain.list_options()
        self.assertTrue('debug_level' in domain_opts.keys())
        self.assertTrue('id_provider' in domain_opts.keys())
        self.assertTrue('auth_provider' in domain_opts.keys())

        del sssdconfig

    def testListProviders(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                srcdir + "/etc/sssd.api.d")

        sssdconfig.new_config()
        junk_domain = sssdconfig.new_domain('junk')
        providers = junk_domain.list_providers()
        self.assertTrue('ldap' in providers.keys())

    def testCreateNewLocalConfig(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                srcdir + "/etc/sssd.api.d")

        sssdconfig.new_config()

        local_domain = sssdconfig.new_domain('LOCAL')
        local_domain.add_provider('local', 'id')
        local_domain.set_option('debug_level', 1)
        local_domain.set_option('default_shell', '/bin/tcsh')
        local_domain.set_active(True)
        sssdconfig.save_domain(local_domain)

        of = '/tmp/testCreateNewLocalConfig.conf'

        #Ensure the output file doesn't exist
        try:
            os.unlink(of)
        except:
            pass

        #Write out the file
        sssdconfig.write(of)

        #Verify that the output file has the correct permissions
        mode = os.stat(of)[ST_MODE]

        #Output files should not be readable or writable by
        #non-owners, and should not be executable by anyone
        self.assertFalse(S_IMODE(mode) & 0177)

        #Remove the output file
        os.unlink(of)


    def testCreateNewLDAPConfig(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                srcdir + "/etc/sssd.api.d")

        sssdconfig.new_config()

        ldap_domain = sssdconfig.new_domain('LDAP')
        ldap_domain.add_provider('ldap', 'id')
        ldap_domain.set_option('debug_level', 1)
        ldap_domain.set_active(True)
        sssdconfig.save_domain(ldap_domain)

        of = '/tmp/testCreateNewLDAPConfig.conf'

        #Ensure the output file doesn't exist
        try:
            os.unlink(of)
        except:
            pass

        #Write out the file
        sssdconfig.write(of)

        #Verify that the output file has the correct permissions
        mode = os.stat(of)[ST_MODE]

        #Output files should not be readable or writable by
        #non-owners, and should not be executable by anyone
        self.assertFalse(S_IMODE(mode) & 0177)

        #Remove the output file
        os.unlink(of)

    def testModifyExistingConfig(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                srcdir + "/etc/sssd.api.d")
        sssdconfig.import_config(srcdir + "/testconfigs/sssd-valid.conf")

        ldap_domain = sssdconfig.get_domain('LDAP')
        ldap_domain.set_option('debug_level', 3)

        ldap_domain.remove_provider('auth')
        ldap_domain.add_provider('krb5', 'auth')
        ldap_domain.set_active(True)
        sssdconfig.save_domain(ldap_domain)

        of = '/tmp/testModifyExistingConfig.conf'

        #Ensure the output file doesn't exist
        try:
            os.unlink(of)
        except:
            pass

        #Write out the file
        sssdconfig.write(of)

        #Verify that the output file has the correct permissions
        mode = os.stat(of)[ST_MODE]

        #Output files should not be readable or writable by
        #non-owners, and should not be executable by anyone
        self.assertFalse(S_IMODE(mode) & 0177)

        #Remove the output file
        os.unlink(of)

    def testSpaces(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")
        sssdconfig.import_config(srcdir + "/testconfigs/sssd-valid.conf")
        ldap_domain = sssdconfig.get_domain('LDAP')
        self.assertEqual(ldap_domain.get_option('auth_provider'), 'ldap')
        self.assertEqual(ldap_domain.get_option('id_provider'), 'ldap')

class SSSDConfigTestInvalid(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        pass

    def testBadBool(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")
        sssdconfig.import_config(srcdir + "/testconfigs/sssd-invalid-badbool.conf")
        self.assertRaises(TypeError,
                          sssdconfig.get_domain,'IPA')

class SSSDConfigTestSSSDService(unittest.TestCase):
    def setUp(self):
        self.schema = SSSDConfig.SSSDConfigSchema(srcdir + "/etc/sssd.api.conf",
                                                  srcdir + "/etc/sssd.api.d")

    def tearDown(self):
        pass

    def testInit(self):
        # Positive test
        service = SSSDConfig.SSSDService('sssd', self.schema)

        # Type Error test
        # Name is not a string
        self.assertRaises(TypeError, SSSDConfig.SSSDService, 3, self.schema)

        # TypeError test
        # schema is not an SSSDSchema
        self.assertRaises(TypeError, SSSDConfig.SSSDService, '3', self)

        # ServiceNotRecognizedError test
        self.assertRaises(SSSDConfig.ServiceNotRecognizedError,
                          SSSDConfig.SSSDService, 'ssd', self.schema)

    def testListOptions(self):
        service = SSSDConfig.SSSDService('sssd', self.schema)

        options = service.list_options()
        control_list = [
            'services',
            'domains',
            'timeout',
            'sbus_timeout',
            're_expression',
            'full_name_format',
            'krb5_rcache_dir',
            'debug_level',
            'debug_timestamps',
            'debug_microseconds',
            'debug_to_files',
            'command',
            'reconnection_retries']

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        self.assertTrue(type(options['reconnection_retries']) == tuple,
                        "Option values should be a tuple")

        self.assertTrue(options['reconnection_retries'][0] == int,
                        "reconnection_retries should require an int. " +
                        "list_options is requiring a %s" %
                        options['reconnection_retries'][0])

        self.assertTrue(options['reconnection_retries'][1] == None,
                        "reconnection_retries should not require a subtype. " +
                        "list_options is requiring a %s" %
                        options['reconnection_retries'][1])

        self.assertTrue(options['reconnection_retries'][3] == None,
                        "reconnection_retries should have no default")

        self.assertTrue(type(options['services']) == tuple,
                        "Option values should be a tuple")

        self.assertTrue(options['services'][0] == list,
                        "services should require an list. " +
                        "list_options is requiring a %s" %
                        options['services'][0])

        self.assertTrue(options['services'][1] == str,
                        "services should require a subtype of str. " +
                        "list_options is requiring a %s" %
                        options['services'][1])

    def testListMandatoryOptions(self):
        service = SSSDConfig.SSSDService('sssd', self.schema)

        options = service.list_mandatory_options()
        control_list = [
            'services',
            'domains']

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        self.assertTrue(type(options['services']) == tuple,
                        "Option values should be a tuple")

        self.assertTrue(options['services'][0] == list,
                        "services should require an list. " +
                        "list_options is requiring a %s" %
                        options['services'][0])

        self.assertTrue(options['services'][1] == str,
                        "services should require a subtype of str. " +
                        "list_options is requiring a %s" %
                        options['services'][1])

    def testSetOption(self):
        service = SSSDConfig.SSSDService('sssd', self.schema)

        # Positive test - Exactly right
        service.set_option('debug_level', 2)
        self.assertEqual(service.get_option('debug_level'), 2)

        # Positive test - Allow converting "safe" values
        service.set_option('debug_level', '2')
        self.assertEqual(service.get_option('debug_level'), 2)

        # Positive test - Remove option if value is None
        service.set_option('debug_level', None)
        self.assertTrue('debug_level' not in service.options.keys())

        # Negative test - Nonexistent Option
        self.assertRaises(SSSDConfig.NoOptionError, service.set_option, 'nosuchoption', 1)

        # Negative test - Incorrect type
        self.assertRaises(TypeError, service.set_option, 'debug_level', 'two')

    def testGetOption(self):
        service = SSSDConfig.SSSDService('sssd', self.schema)

        # Positive test - Single-valued
        self.assertEqual(service.get_option('config_file_version'), 2)

        # Positive test - List of values
        self.assertEqual(service.get_option('services'), ['nss', 'pam'])

        # Negative Test - Bad Option
        self.assertRaises(SSSDConfig.NoOptionError, service.get_option, 'nosuchoption')

    def testGetAllOptions(self):
        service = SSSDConfig.SSSDService('sssd', self.schema)

        #Positive test
        options = service.get_all_options()
        control_list = [
            'config_file_version',
            'services']

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

    def testRemoveOption(self):
        service = SSSDConfig.SSSDService('sssd', self.schema)

        # Positive test - Remove an option that exists
        self.assertEqual(service.get_option('services'), ['nss', 'pam'])
        service.remove_option('services')
        self.assertRaises(SSSDConfig.NoOptionError, service.get_option, 'debug_level')

        # Positive test - Remove an option that doesn't exist
        self.assertRaises(SSSDConfig.NoOptionError, service.get_option, 'nosuchentry')
        service.remove_option('nosuchentry')

class SSSDConfigTestSSSDDomain(unittest.TestCase):
    def setUp(self):
        self.schema = SSSDConfig.SSSDConfigSchema(srcdir + "/etc/sssd.api.conf",
                                                  srcdir + "/etc/sssd.api.d")

    def tearDown(self):
        pass

    def testInit(self):
        # Positive Test
        domain = SSSDConfig.SSSDDomain('mydomain', self.schema)

        # Negative Test - Name not a string
        self.assertRaises(TypeError, SSSDConfig.SSSDDomain, 2, self.schema)

        # Negative Test - Schema is not an SSSDSchema
        self.assertRaises(TypeError, SSSDConfig.SSSDDomain, 'mydomain', self)

    def testGetName(self):
        # Positive Test
        domain = SSSDConfig.SSSDDomain('mydomain', self.schema)

        self.assertEqual(domain.get_name(), 'mydomain')

    def testSetActive(self):
        #Positive Test
        domain = SSSDConfig.SSSDDomain('mydomain', self.schema)

        # Should default to inactive
        self.assertFalse(domain.active)
        domain.set_active(True)
        self.assertTrue(domain.active)
        domain.set_active(False)
        self.assertFalse(domain.active)

    def testListOptions(self):
        domain = SSSDConfig.SSSDDomain('sssd', self.schema)

        # First test default options
        options = domain.list_options()
        control_list = [
            'description',
            'debug_level',
            'debug_timestamps',
            'min_id',
            'max_id',
            'timeout',
            'try_inotify',
            'command',
            'enumerate',
            'cache_credentials',
            'store_legacy_passwords',
            'use_fully_qualified_names',
            'filter_users',
            'filter_groups',
            'entry_cache_timeout',
            'entry_cache_user_timeout',
            'entry_cache_group_timeout',
            'entry_cache_netgroup_timeout',
            'entry_cache_service_timeout',
            'entry_cache_autofs_timeout',
            'lookup_family_order',
            'account_cache_expiration',
            'dns_resolver_timeout',
            'dns_discovery_domain',
            'override_gid',
            'case_sensitive',
            'override_homedir',
            'id_provider',
            'auth_provider',
            'access_provider',
            'chpass_provider',
            'sudo_provider',
            'autofs_provider',
            'session_provider',
            'hostid_provider']

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        self.assertTrue(type(options['max_id']) == tuple,
                        "Option values should be a tuple")

        self.assertTrue(options['max_id'][0] == int,
                        "max_id should require an int. " +
                        "list_options is requiring a %s" %
                        options['max_id'][0])

        self.assertTrue(options['max_id'][1] == None,
                        "max_id should not require a subtype. " +
                        "list_options is requiring a %s" %
                        options['max_id'][1])

        # Add a provider and verify that the new options appear
        domain.add_provider('local', 'id')
        control_list.extend(
            ['default_shell',
             'base_directory',
             'create_homedir',
             'remove_homedir',
             'homedir_umask',
             'skel_dir',
             'mail_dir',
             'userdel_cmd'])

        options = domain.list_options()

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        # Add a provider that has global options and verify that
        # The new options appear.
        domain.add_provider('krb5', 'auth')

        backup_list = control_list[:]
        control_list.extend(
            ['krb5_server',
             'krb5_realm',
             'krb5_kpasswd',
             'krb5_ccachedir',
             'krb5_ccname_template',
             'krb5_keytab',
             'krb5_validate',
             'krb5_store_password_if_offline',
             'krb5_auth_timeout',
             'krb5_renewable_lifetime',
             'krb5_lifetime',
             'krb5_renew_interval',
             'krb5_use_fast',
             'krb5_fast_principal',
             'krb5_canonicalize'])

        options = domain.list_options()

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        control_list.extend(['krb5_kdcip'])

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        # Remove the auth domain and verify that the options
        # revert to the backup_list
        domain.remove_provider('auth')
        options = domain.list_options()

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in backup_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in backup_list,
                            'Option [%s] unexpectedly found' %
                            option)

    def testListMandatoryOptions(self):
        domain = SSSDConfig.SSSDDomain('sssd', self.schema)

        # First test default options
        options = domain.list_mandatory_options()
        control_list = ['id_provider']

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        # Add a provider that has global options and verify that
        # The new options appear.
        domain.add_provider('krb5', 'auth')

        backup_list = control_list[:]
        control_list.extend(['krb5_realm'])

        options = domain.list_mandatory_options()

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        # Remove the auth domain and verify that the options
        # revert to the backup_list
        domain.remove_provider('auth')
        options = domain.list_mandatory_options()

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in backup_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in backup_list,
                            'Option [%s] unexpectedly found' %
                            option)

    def testListProviders(self):
        domain = SSSDConfig.SSSDDomain('sssd', self.schema)

        control_provider_dict = {
            'ipa': ['id', 'auth', 'access', 'chpass', 'autofs' ],
            'local': ['id', 'auth', 'chpass'],
            'ldap': ['id', 'auth', 'access', 'chpass', 'sudo', 'autofs'],
            'krb5': ['auth', 'access', 'chpass'],
            'proxy': ['id', 'auth'],
            'simple': ['access'],
            'permit': ['access'],
            'deny': ['access']}

        providers = domain.list_providers()

        # Ensure that all of the expected defaults are there
        for provider in control_provider_dict.keys():
            for ptype in control_provider_dict[provider]:
                self.assertTrue(providers.has_key(provider))
                self.assertTrue(ptype in providers[provider])

        for provider in providers.keys():
            for ptype in providers[provider]:
                self.assertTrue(control_provider_dict.has_key(provider))
                self.assertTrue(ptype in control_provider_dict[provider])

    def testListProviderOptions(self):
        domain = SSSDConfig.SSSDDomain('sssd', self.schema)

        # Test looking up a specific provider type
        options = domain.list_provider_options('krb5', 'auth')
        control_list = [
            'krb5_server',
            'krb5_kdcip',
            'krb5_realm',
            'krb5_kpasswd',
            'krb5_ccachedir',
            'krb5_ccname_template',
            'krb5_keytab',
            'krb5_validate',
            'krb5_store_password_if_offline',
            'krb5_auth_timeout',
            'krb5_renewable_lifetime',
            'krb5_lifetime',
            'krb5_renew_interval',
            'krb5_use_fast',
            'krb5_fast_principal',
            'krb5_canonicalize']

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        #Test looking up all provider values
        options = domain.list_provider_options('krb5')
        control_list.extend(['krb5_kpasswd'])

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

    def testAddProvider(self):
        domain = SSSDConfig.SSSDDomain('sssd', self.schema)

        # Positive Test
        domain.add_provider('local', 'id')

        # Negative Test - No such backend type
        self.assertRaises(SSSDConfig.NoSuchProviderError,
                          domain.add_provider, 'nosuchbackend', 'auth')

        # Negative Test - No such backend subtype
        self.assertRaises(SSSDConfig.NoSuchProviderSubtypeError,
                          domain.add_provider, 'ldap', 'nosuchsubtype')

        # Negative Test - Try to add a second provider of the same type
        self.assertRaises(SSSDConfig.ProviderSubtypeInUse,
                          domain.add_provider, 'ldap', 'id')

    def testRemoveProvider(self):
        domain = SSSDConfig.SSSDDomain('sssd', self.schema)

        # First test default options
        options = domain.list_options()
        control_list = [
            'description',
            'debug_level',
            'debug_timestamps',
            'min_id',
            'max_id',
            'timeout',
            'try_inotify',
            'command',
            'enumerate',
            'cache_credentials',
            'store_legacy_passwords',
            'use_fully_qualified_names',
            'filter_users',
            'filter_groups',
            'entry_cache_timeout',
            'entry_cache_user_timeout',
            'entry_cache_group_timeout',
            'entry_cache_netgroup_timeout',
            'entry_cache_service_timeout',
            'entry_cache_autofs_timeout',
            'account_cache_expiration',
            'lookup_family_order',
            'dns_resolver_timeout',
            'dns_discovery_domain',
            'override_gid',
            'case_sensitive',
            'override_homedir',
            'id_provider',
            'auth_provider',
            'access_provider',
            'chpass_provider',
            'sudo_provider',
            'autofs_provider',
            'session_provider',
            'hostid_provider']

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        self.assertTrue(type(options['max_id']) == tuple,
                        "Option values should be a tuple")

        self.assertTrue(options['max_id'][0] == int,
                        "config_file_version should require an int. " +
                        "list_options is requiring a %s" %
                        options['max_id'][0])

        self.assertTrue(options['max_id'][1] == None,
                        "config_file_version should not require a subtype. " +
                        "list_options is requiring a %s" %
                        options['max_id'][1])

        # Add a provider and verify that the new options appear
        domain.add_provider('local', 'id')
        control_list.extend(
            ['default_shell',
             'base_directory',
             'create_homedir',
             'remove_homedir',
             'homedir_umask',
             'skel_dir',
             'mail_dir',
             'userdel_cmd'])

        options = domain.list_options()

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        # Add a provider that has global options and verify that
        # The new options appear.
        domain.add_provider('krb5', 'auth')

        backup_list = control_list[:]
        control_list.extend(
            ['krb5_server',
             'krb5_kdcip',
             'krb5_realm',
             'krb5_kpasswd',
             'krb5_ccachedir',
             'krb5_ccname_template',
             'krb5_keytab',
             'krb5_validate',
             'krb5_store_password_if_offline',
             'krb5_auth_timeout',
             'krb5_renewable_lifetime',
             'krb5_lifetime',
             'krb5_renew_interval',
             'krb5_use_fast',
             'krb5_fast_principal',
             'krb5_canonicalize'])

        options = domain.list_options()

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in control_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in control_list,
                            'Option [%s] unexpectedly found' %
                            option)

        # Remove the local ID provider and add an LDAP one
        # LDAP ID providers can also use the krb5_realm
        domain.remove_provider('id')

        domain.add_provider('ldap', 'id')

        # Set the krb5_realm option and the ldap_uri option
        domain.set_option('krb5_realm', 'EXAMPLE.COM')
        domain.set_option('ldap_uri', 'ldap://ldap.example.com')

        self.assertEquals(domain.get_option('krb5_realm'),
                          'EXAMPLE.COM')
        self.assertEquals(domain.get_option('ldap_uri'),
                          'ldap://ldap.example.com')

        # Remove the LDAP provider and verify that krb5_realm remains
        domain.remove_provider('id')
        self.assertEquals(domain.get_option('krb5_realm'),
                  'EXAMPLE.COM')
        self.assertFalse(domain.options.has_key('ldap_uri'))

        # Put the LOCAL provider back
        domain.add_provider('local', 'id')

        # Remove the auth domain and verify that the options
        # revert to the backup_list
        domain.remove_provider('auth')
        options = domain.list_options()

        self.assertTrue(type(options) == dict,
                        "Options should be a dictionary")

        # Ensure that all of the expected defaults are there
        for option in backup_list:
            self.assertTrue(option in options.keys(),
                            "Option [%s] missing" %
                            option)

        # Ensure that there aren't any unexpected options listed
        for option in options.keys():
            self.assertTrue(option in backup_list,
                            'Option [%s] unexpectedly found' %
                            option)

        # Ensure that the krb5_realm option is now gone
        self.assertFalse(domain.options.has_key('krb5_realm'))

        # Test removing nonexistent provider - Real
        domain.remove_provider('id')

        # Test removing nonexistent provider - Bad backend type
        # Should pass without complaint
        domain.remove_provider('id')

        # Test removing nonexistent provider - Bad provider type
        # Should pass without complaint
        domain.remove_provider('nosuchprovider')

    def testGetOption(self):
        domain = SSSDConfig.SSSDDomain('sssd', self.schema)

        # Negative Test - Try to get valid option that is not set
        self.assertRaises(SSSDConfig.NoOptionError, domain.get_option, 'max_id')

        # Positive Test - Set the above option and get it
        domain.set_option('max_id', 10000)
        self.assertEqual(domain.get_option('max_id'), 10000)

        # Negative Test - Try yo get invalid option
        self.assertRaises(SSSDConfig.NoOptionError, domain.get_option, 'nosuchoption')

    def testSetOption(self):
        domain = SSSDConfig.SSSDDomain('sssd', self.schema)

        # Positive Test
        domain.set_option('max_id', 10000)
        self.assertEqual(domain.get_option('max_id'), 10000)

        # Positive Test - Remove option if value is None
        domain.set_option('max_id', None)
        self.assertTrue('max_id' not in domain.get_all_options().keys())

        # Negative Test - invalid option
        self.assertRaises(SSSDConfig.NoOptionError, domain.set_option, 'nosuchoption', 1)

        # Negative Test - incorrect type
        self.assertRaises(TypeError, domain.set_option, 'max_id', 'a string')

        # Positive Test - Coax options to appropriate type
        domain.set_option('max_id', '10000')
        self.assertEqual(domain.get_option('max_id'), 10000)

        domain.set_option('max_id', 30.2)
        self.assertEqual(domain.get_option('max_id'), 30)

    def testRemoveOption(self):
        domain = SSSDConfig.SSSDDomain('sssd', self.schema)

        # Positive test - Remove unset but valid option
        self.assertFalse('max_id' in domain.get_all_options().keys())
        domain.remove_option('max_id')
        self.assertFalse('max_id' in domain.get_all_options().keys())

        # Positive test - Remove unset and unknown option
        self.assertFalse('nosuchoption' in domain.get_all_options().keys())
        domain.remove_option('nosuchoption')
        self.assertFalse('nosuchoption' in domain.get_all_options().keys())

    def testSetName(self):
        domain = SSSDConfig.SSSDDomain('sssd', self.schema)

        # Positive test - Change the name once
        domain.set_name('sssd2');
        self.assertEqual(domain.get_name(), 'sssd2')
        self.assertEqual(domain.oldname, 'sssd')

        # Positive test - Change the name a second time
        domain.set_name('sssd3')
        self.assertEqual(domain.get_name(), 'sssd3')
        self.assertEqual(domain.oldname, 'sssd')

        # Negative test - try setting the name to a non-string
        self.assertRaises(TypeError,
                          domain.set_name, 4)

class SSSDConfigTestSSSDConfig(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        pass

    def testInit(self):
        # Positive test
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - No Such File
        self.assertRaises(IOError,
                          SSSDConfig.SSSDConfig, "nosuchfile.api.conf", srcdir + "/etc/sssd.api.d")

        # Negative Test - Schema is not parsable
        self.assertRaises(SSSDConfig.ParsingError,
                          SSSDConfig.SSSDConfig, srcdir + "/testconfigs/noparse.api.conf", srcdir + "/etc/sssd.api.d")

    def testImportConfig(self):
        # Positive Test
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")
        sssdconfig.import_config(srcdir + "/testconfigs/sssd-valid.conf")

        # Verify that all sections were imported
        control_list = [
            'sssd',
            'nss',
            'pam',
            'domain/PROXY',
            'domain/IPA',
            'domain/LOCAL',
            'domain/LDAP',
            'domain/INVALIDPROVIDER',
            'domain/INVALIDOPTION',
            ]

        for section in control_list:
            self.assertTrue(sssdconfig.has_section(section),
                            "Section [%s] missing" %
                            section)
        for section in sssdconfig.sections():
            self.assertTrue(section['name'] in control_list)

        # Verify that all options were imported for a section
        control_list = [
            'services',
            'reconnection_retries',
            'domains',
            'debug_timestamps',
            'config_file_version']

        for option in control_list:
            self.assertTrue(sssdconfig.has_option('sssd', option),
                            "Option [%s] missing from [sssd]" %
                            option)
        for option in sssdconfig.options('sssd'):
            if option['type'] in ('empty', 'comment'):
                continue
            self.assertTrue(option['name'] in control_list,
                            "Option [%s] unexpectedly found" %
                            option)

        #TODO: Check the types and values of the settings

        # Negative Test - Missing config file
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")
        self.assertRaises(IOError, sssdconfig.import_config, "nosuchfile.conf")

        # Negative Test - Invalid config file
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")
        self.assertRaises(SSSDConfig.ParsingError, sssdconfig.import_config, srcdir + "/testconfigs/sssd-invalid.conf")

        # Negative Test - Invalid config file version
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")
        self.assertRaises(SSSDConfig.ParsingError, sssdconfig.import_config, srcdir + "/testconfigs/sssd-badversion.conf")

        # Negative Test - No config file version
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")
        self.assertRaises(SSSDConfig.ParsingError, sssdconfig.import_config, srcdir + "/testconfigs/sssd-noversion.conf")

        # Negative Test - Already initialized
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")
        sssdconfig.import_config(srcdir + "/testconfigs/sssd-valid.conf")
        self.assertRaises(SSSDConfig.AlreadyInitializedError,
                          sssdconfig.import_config, srcdir + "/testconfigs/sssd-valid.conf")

    def testNewConfig(self):
        # Positive Test
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")
        sssdconfig.new_config()

        # Check that the defaults were set
        control_list = [
            'sssd',
            'nss',
            'pam',
            'sudo',
            'autofs']
        for section in control_list:
            self.assertTrue(sssdconfig.has_section(section),
                            "Section [%s] missing" %
                            section)
        for section in sssdconfig.sections():
            self.assertTrue(section['name'] in control_list)

        control_list = [
            'config_file_version',
            'services']
        for option in control_list:
            self.assertTrue(sssdconfig.has_option('sssd', option),
                            "Option [%s] missing from [sssd]" %
                            option)
        for option in sssdconfig.options('sssd'):
            if option['type'] in ('empty', 'comment'):
                continue
            self.assertTrue(option['name'] in control_list,
                            "Option [%s] unexpectedly found" %
                            option)

        # Negative Test - Already Initialized
        self.assertRaises(SSSDConfig.AlreadyInitializedError, sssdconfig.new_config)

    def testWrite(self):
        #TODO Write tests to compare output files
        pass

    def testListServices(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - sssdconfig not initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.list_services)

        sssdconfig.new_config()

        control_list = [
            'sssd',
            'pam',
            'nss',
            'sudo',
            'autofs' ]
        service_list = sssdconfig.list_services()
        for service in control_list:
            self.assertTrue(service in service_list,
                            "Service [%s] missing" %
                            service)
        for service in service_list:
            self.assertTrue(service in control_list,
                            "Service [%s] unexpectedly found" %
                            service)

    def testGetService(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - Not initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.get_service, 'sssd')

        sssdconfig.import_config(srcdir + '/testconfigs/sssd-valid.conf')

        service = sssdconfig.get_service('sssd')
        self.assertTrue(isinstance(service, SSSDConfig.SSSDService))

        # Verify the contents of this service
        self.assertEqual(type(service.get_option('debug_timestamps')), bool)
        self.assertFalse(service.get_option('debug_timestamps'))

        # Negative Test - No such service
        self.assertRaises(SSSDConfig.NoServiceError, sssdconfig.get_service, 'nosuchservice')

        # Positive test - Service with invalid option loads
        # but ignores the invalid option
        service = sssdconfig.get_service('pam')
        self.assertFalse(service.options.has_key('nosuchoption'))

    def testNewService(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - Not initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.new_service, 'sssd')

        sssdconfig.new_config()

        # Positive Test
        # First need to remove the existing service
        sssdconfig.delete_service('sssd')
        service = sssdconfig.new_service('sssd')
        self.failUnless(service.get_name() in sssdconfig.list_services())

        # TODO: check that the values of this new service
        # are set to the defaults from the schema

    def testDeleteService(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - Not initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.delete_service, 'sssd')

        sssdconfig.new_config()

        # Positive Test
        service = sssdconfig.delete_service('sssd')

    def testSaveService(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        new_service = SSSDConfig.SSSDService('sssd', sssdconfig.schema)

        # Negative Test - Not initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.save_service, new_service)

        # Positive Test
        sssdconfig.new_config()
        sssdconfig.save_service(new_service)

        # TODO: check that all entries were saved correctly (change a few)

        # Negative Test - Type Error
        self.assertRaises(TypeError, sssdconfig.save_service, self)

    def testListActiveDomains(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - Not Initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.list_active_domains)

        # Positive Test
        sssdconfig.import_config(srcdir + '/testconfigs/sssd-valid.conf')

        control_list = [
            'IPA',
            'LOCAL']
        active_domains = sssdconfig.list_active_domains()

        for domain in control_list:
            self.assertTrue(domain in active_domains,
                            "Domain [%s] missing" %
                            domain)
        for domain in active_domains:
            self.assertTrue(domain in control_list,
                            "Domain [%s] unexpectedly found" %
                            domain)

    def testListInactiveDomains(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - Not Initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.list_inactive_domains)

        # Positive Test
        sssdconfig.import_config(srcdir + '/testconfigs/sssd-valid.conf')

        control_list = [
            'PROXY',
            'LDAP',
            'INVALIDPROVIDER',
            'INVALIDOPTION',
            ]
        inactive_domains = sssdconfig.list_inactive_domains()

        for domain in control_list:
            self.assertTrue(domain in inactive_domains,
                            "Domain [%s] missing" %
                            domain)
        for domain in inactive_domains:
            self.assertTrue(domain in control_list,
                            "Domain [%s] unexpectedly found" %
                            domain)

    def testListDomains(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - Not Initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.list_domains)

        # Positive Test
        sssdconfig.import_config(srcdir + '/testconfigs/sssd-valid.conf')

        control_list = [
            'IPA',
            'LOCAL',
            'PROXY',
            'LDAP',
            'INVALIDPROVIDER',
            'INVALIDOPTION',
            ]
        domains = sssdconfig.list_domains()

        for domain in control_list:
            self.assertTrue(domain in domains,
                            "Domain [%s] missing" %
                            domain)
        for domain in domains:
            self.assertTrue(domain in control_list,
                            "Domain [%s] unexpectedly found" %
                            domain)

    def testGetDomain(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - Not initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.get_domain, 'sssd')

        sssdconfig.import_config(srcdir + '/testconfigs/sssd-valid.conf')

        domain = sssdconfig.get_domain('IPA')
        self.assertTrue(isinstance(domain, SSSDConfig.SSSDDomain))
        self.assertTrue(domain.active)

        domain = sssdconfig.get_domain('LDAP')
        self.assertTrue(isinstance(domain, SSSDConfig.SSSDDomain))
        self.assertFalse(domain.active)

        # TODO verify the contents of this domain
        self.assertTrue(domain.get_option('ldap_id_use_start_tls'))

        # Negative Test - No such domain
        self.assertRaises(SSSDConfig.NoDomainError, sssdconfig.get_domain, 'nosuchdomain')

        # Positive Test - Domain with unknown provider
        # Expected result: Domain is imported, but does not contain the
        # unknown provider entry
        domain = sssdconfig.get_domain('INVALIDPROVIDER')
        self.assertFalse(domain.options.has_key('chpass_provider'))

        # Positive Test - Domain with unknown option
        # Expected result: Domain is imported, but does not contain the
        # unknown option entry
        domain = sssdconfig.get_domain('INVALIDOPTION')
        self.assertFalse(domain.options.has_key('nosuchoption'))

    def testNewDomain(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - Not initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.new_domain, 'example.com')

        sssdconfig.new_config()

        # Positive Test
        domain = sssdconfig.new_domain('example.com')
        self.assertTrue(isinstance(domain, SSSDConfig.SSSDDomain))
        self.failUnless(domain.get_name() in sssdconfig.list_domains())
        self.failUnless(domain.get_name() in sssdconfig.list_inactive_domains())

        # TODO: check that the values of this new domain
        # are set to the defaults from the schema

    def testDeleteDomain(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")

        # Negative Test - Not initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.delete_domain, 'IPA')

        # Positive Test
        sssdconfig.import_config(srcdir + '/testconfigs/sssd-valid.conf')

        self.assertTrue('IPA' in sssdconfig.list_domains())
        self.assertTrue('IPA' in sssdconfig.list_active_domains())
        self.assertTrue(sssdconfig.has_section('domain/IPA'))
        sssdconfig.delete_domain('IPA')
        self.assertFalse('IPA' in sssdconfig.list_domains())
        self.assertFalse('IPA' in sssdconfig.list_active_domains())
        self.assertFalse(sssdconfig.has_section('domain/IPA'))

    def testSaveDomain(self):
        sssdconfig = SSSDConfig.SSSDConfig(srcdir + "/etc/sssd.api.conf",
                                           srcdir + "/etc/sssd.api.d")
        # Negative Test - Not initialized
        self.assertRaises(SSSDConfig.NotInitializedError, sssdconfig.save_domain, 'IPA')

        # Positive Test
        sssdconfig.new_config()
        domain = sssdconfig.new_domain('example.com')
        domain.add_provider('ldap', 'id')
        domain.set_option('ldap_uri', 'ldap://ldap.example.com')
        domain.set_active(True)
        sssdconfig.save_domain(domain)

        self.assertTrue('example.com' in sssdconfig.list_domains())
        self.assertTrue('example.com' in sssdconfig.list_active_domains())
        self.assertEqual(sssdconfig.get('domain/example.com', 'ldap_uri'),
                         'ldap://ldap.example.com')

        # Negative Test - Type Error
        self.assertRaises(TypeError, sssdconfig.save_domain, self)

        # Positive test - Change the domain name and save it
        domain.set_name('example.com2')
        self.assertEqual(domain.name,'example.com2')
        self.assertEqual(domain.oldname,'example.com')
        sssdconfig.save_domain(domain)

        self.assertTrue('example.com2' in sssdconfig.list_domains())
        self.assertTrue('example.com2' in sssdconfig.list_active_domains())
        self.assertTrue(sssdconfig.has_section('domain/example.com2'))
        self.assertEqual(sssdconfig.get('domain/example.com2',
                                        'ldap_uri'),
                         'ldap://ldap.example.com')
        self.assertFalse('example.com' in sssdconfig.list_domains())
        self.assertFalse('example.com' in sssdconfig.list_active_domains())
        self.assertFalse('example.com' in sssdconfig.list_inactive_domains())
        self.assertFalse(sssdconfig.has_section('domain/example.com'))
        self.assertEquals(domain.oldname, None)

        # Positive test - Set the domain inactive and save it
        activelist = sssdconfig.list_active_domains()
        inactivelist = sssdconfig.list_inactive_domains()

        domain.set_active(False)
        sssdconfig.save_domain(domain)

        self.assertFalse('example.com2' in sssdconfig.list_active_domains())
        self.assertTrue('example.com2' in sssdconfig.list_inactive_domains())

        self.assertEquals(len(sssdconfig.list_active_domains()),
                          len(activelist)-1)
        self.assertEquals(len(sssdconfig.list_inactive_domains()),
                          len(inactivelist)+1)

        # Positive test - Set the domain active and save it
        activelist = sssdconfig.list_active_domains()
        inactivelist = sssdconfig.list_inactive_domains()
        domain.set_active(True)
        sssdconfig.save_domain(domain)

        self.assertTrue('example.com2' in sssdconfig.list_active_domains())
        self.assertFalse('example.com2' in sssdconfig.list_inactive_domains())

        self.assertEquals(len(sssdconfig.list_active_domains()),
                          len(activelist)+1)
        self.assertEquals(len(sssdconfig.list_inactive_domains()),
                          len(inactivelist)-1)

        # Positive test - Set the domain inactive and save it
        activelist = sssdconfig.list_active_domains()
        inactivelist = sssdconfig.list_inactive_domains()

        sssdconfig.deactivate_domain(domain.get_name())

        self.assertFalse('example.com2' in sssdconfig.list_active_domains())
        self.assertTrue('example.com2' in sssdconfig.list_inactive_domains())

        self.assertEquals(len(sssdconfig.list_active_domains()),
                          len(activelist)-1)
        self.assertEquals(len(sssdconfig.list_inactive_domains()),
                          len(inactivelist)+1)

        # Positive test - Set the domain active and save it
        activelist = sssdconfig.list_active_domains()
        inactivelist = sssdconfig.list_inactive_domains()

        sssdconfig.activate_domain(domain.get_name())