package plugin import ( "bufio" "context" "crypto/subtle" "crypto/tls" "errors" "fmt" "hash" "io" "io/ioutil" "net" "os" "os/exec" "path/filepath" "strconv" "strings" "sync" "sync/atomic" "time" "unicode" hclog "github.com/hashicorp/go-hclog" ) // If this is 1, then we've called CleanupClients. This can be used // by plugin RPC implementations to change error behavior since you // can expected network connection errors at this point. This should be // read by using sync/atomic. var Killed uint32 = 0 // This is a slice of the "managed" clients which are cleaned up when // calling Cleanup var managedClients = make([]*Client, 0, 5) var managedClientsLock sync.Mutex // Error types var ( // ErrProcessNotFound is returned when a client is instantiated to // reattach to an existing process and it isn't found. ErrProcessNotFound = errors.New("Reattachment process not found") // ErrChecksumsDoNotMatch is returned when binary's checksum doesn't match // the one provided in the SecureConfig. ErrChecksumsDoNotMatch = errors.New("checksums did not match") // ErrSecureNoChecksum is returned when an empty checksum is provided to the // SecureConfig. ErrSecureConfigNoChecksum = errors.New("no checksum provided") // ErrSecureNoHash is returned when a nil Hash object is provided to the // SecureConfig. ErrSecureConfigNoHash = errors.New("no hash implementation provided") // ErrSecureConfigAndReattach is returned when both Reattach and // SecureConfig are set. ErrSecureConfigAndReattach = errors.New("only one of Reattach or SecureConfig can be set") ) // Client handles the lifecycle of a plugin application. It launches // plugins, connects to them, dispenses interface implementations, and handles // killing the process. // // Plugin hosts should use one Client for each plugin executable. To // dispense a plugin type, use the `Client.Client` function, and then // cal `Dispense`. This awkward API is mostly historical but is used to split // the client that deals with subprocess management and the client that // does RPC management. // // See NewClient and ClientConfig for using a Client. type Client struct { config *ClientConfig exited bool doneLogging chan struct{} l sync.Mutex address net.Addr process *os.Process client ClientProtocol protocol Protocol logger hclog.Logger doneCtx context.Context } // ClientConfig is the configuration used to initialize a new // plugin client. After being used to initialize a plugin client, // that configuration must not be modified again. type ClientConfig struct { // HandshakeConfig is the configuration that must match servers. HandshakeConfig // Plugins are the plugins that can be consumed. Plugins map[string]Plugin // One of the following must be set, but not both. // // Cmd is the unstarted subprocess for starting the plugin. If this is // set, then the Client starts the plugin process on its own and connects // to it. // // Reattach is configuration for reattaching to an existing plugin process // that is already running. This isn't common. Cmd *exec.Cmd Reattach *ReattachConfig // SecureConfig is configuration for verifying the integrity of the // executable. It can not be used with Reattach. SecureConfig *SecureConfig // TLSConfig is used to enable TLS on the RPC client. TLSConfig *tls.Config // Managed represents if the client should be managed by the // plugin package or not. If true, then by calling CleanupClients, // it will automatically be cleaned up. Otherwise, the client // user is fully responsible for making sure to Kill all plugin // clients. By default the client is _not_ managed. Managed bool // The minimum and maximum port to use for communicating with // the subprocess. If not set, this defaults to 10,000 and 25,000 // respectively. MinPort, MaxPort uint // StartTimeout is the timeout to wait for the plugin to say it // has started successfully. StartTimeout time.Duration // If non-nil, then the stderr of the client will be written to here // (as well as the log). This is the original os.Stderr of the subprocess. // This isn't the output of synced stderr. Stderr io.Writer // SyncStdout, SyncStderr can be set to override the // respective os.Std* values in the plugin. Care should be taken to // avoid races here. If these are nil, then this will automatically be // hooked up to os.Stdin, Stdout, and Stderr, respectively. // // If the default values (nil) are used, then this package will not // sync any of these streams. SyncStdout io.Writer SyncStderr io.Writer // AllowedProtocols is a list of allowed protocols. If this isn't set, // then only netrpc is allowed. This is so that older go-plugin systems // can show friendly errors if they see a plugin with an unknown // protocol. // // By setting this, you can cause an error immediately on plugin start // if an unsupported protocol is used with a good error message. // // If this isn't set at all (nil value), then only net/rpc is accepted. // This is done for legacy reasons. You must explicitly opt-in to // new protocols. AllowedProtocols []Protocol // Logger is the logger that the client will used. If none is provided, // it will default to hclog's default logger. Logger hclog.Logger } // ReattachConfig is used to configure a client to reattach to an // already-running plugin process. You can retrieve this information by // calling ReattachConfig on Client. type ReattachConfig struct { Protocol Protocol Addr net.Addr Pid int } // SecureConfig is used to configure a client to verify the integrity of an // executable before running. It does this by verifying the checksum is // expected. Hash is used to specify the hashing method to use when checksumming // the file. The configuration is verified by the client by calling the // SecureConfig.Check() function. // // The host process should ensure the checksum was provided by a trusted and // authoritative source. The binary should be installed in such a way that it // can not be modified by an unauthorized user between the time of this check // and the time of execution. type SecureConfig struct { Checksum []byte Hash hash.Hash } // Check takes the filepath to an executable and returns true if the checksum of // the file matches the checksum provided in the SecureConfig. func (s *SecureConfig) Check(filePath string) (bool, error) { if len(s.Checksum) == 0 { return false, ErrSecureConfigNoChecksum } if s.Hash == nil { return false, ErrSecureConfigNoHash } file, err := os.Open(filePath) if err != nil { return false, err } defer file.Close() _, err = io.Copy(s.Hash, file) if err != nil { return false, err } sum := s.Hash.Sum(nil) return subtle.ConstantTimeCompare(sum, s.Checksum) == 1, nil } // This makes sure all the managed subprocesses are killed and properly // logged. This should be called before the parent process running the // plugins exits. // // This must only be called _once_. func CleanupClients() { // Set the killed to true so that we don't get unexpected panics atomic.StoreUint32(&Killed, 1) // Kill all the managed clients in parallel and use a WaitGroup // to wait for them all to finish up. var wg sync.WaitGroup managedClientsLock.Lock() for _, client := range managedClients { wg.Add(1) go func(client *Client) { client.Kill() wg.Done() }(client) } managedClientsLock.Unlock() wg.Wait() } // Creates a new plugin client which manages the lifecycle of an external // plugin and gets the address for the RPC connection. // // The client must be cleaned up at some point by calling Kill(). If // the client is a managed client (created with NewManagedClient) you // can just call CleanupClients at the end of your program and they will // be properly cleaned. func NewClient(config *ClientConfig) (c *Client) { if config.MinPort == 0 && config.MaxPort == 0 { config.MinPort = 10000 config.MaxPort = 25000 } if config.StartTimeout == 0 { config.StartTimeout = 1 * time.Minute } if config.Stderr == nil { config.Stderr = ioutil.Discard } if config.SyncStdout == nil { config.SyncStdout = ioutil.Discard } if config.SyncStderr == nil { config.SyncStderr = ioutil.Discard } if config.AllowedProtocols == nil { config.AllowedProtocols = []Protocol{ProtocolNetRPC} } if config.Logger == nil { config.Logger = hclog.New(&hclog.LoggerOptions{ Output: hclog.DefaultOutput, Level: hclog.Trace, Name: "plugin", }) } c = &Client{ config: config, logger: config.Logger, } if config.Managed { managedClientsLock.Lock() managedClients = append(managedClients, c) managedClientsLock.Unlock() } return } // Client returns the protocol client for this connection. // // Subsequent calls to this will return the same client. func (c *Client) Client() (ClientProtocol, error) { _, err := c.Start() if err != nil { return nil, err } c.l.Lock() defer c.l.Unlock() if c.client != nil { return c.client, nil } switch c.protocol { case ProtocolNetRPC: c.client, err = newRPCClient(c) case ProtocolGRPC: c.client, err = newGRPCClient(c.doneCtx, c) default: return nil, fmt.Errorf("unknown server protocol: %s", c.protocol) } if err != nil { c.client = nil return nil, err } return c.client, nil } // Tells whether or not the underlying process has exited. func (c *Client) Exited() bool { c.l.Lock() defer c.l.Unlock() return c.exited } // End the executing subprocess (if it is running) and perform any cleanup // tasks necessary such as capturing any remaining logs and so on. // // This method blocks until the process successfully exits. // // This method can safely be called multiple times. func (c *Client) Kill() { // Grab a lock to read some private fields. c.l.Lock() process := c.process addr := c.address doneCh := c.doneLogging c.l.Unlock() // If there is no process, we never started anything. Nothing to kill. if process == nil { return } // We need to check for address here. It is possible that the plugin // started (process != nil) but has no address (addr == nil) if the // plugin failed at startup. If we do have an address, we need to close // the plugin net connections. graceful := false if addr != nil { // Close the client to cleanly exit the process. client, err := c.Client() if err == nil { err = client.Close() // If there is no error, then we attempt to wait for a graceful // exit. If there was an error, we assume that graceful cleanup // won't happen and just force kill. graceful = err == nil if err != nil { // If there was an error just log it. We're going to force // kill in a moment anyways. c.logger.Warn("error closing client during Kill", "err", err) } } } // If we're attempting a graceful exit, then we wait for a short period // of time to allow that to happen. To wait for this we just wait on the // doneCh which would be closed if the process exits. if graceful { select { case <-doneCh: return case <-time.After(250 * time.Millisecond): } } // If graceful exiting failed, just kill it process.Kill() // Wait for the client to finish logging so we have a complete log <-doneCh } // Starts the underlying subprocess, communicating with it to negotiate // a port for RPC connections, and returning the address to connect via RPC. // // This method is safe to call multiple times. Subsequent calls have no effect. // Once a client has been started once, it cannot be started again, even if // it was killed. func (c *Client) Start() (addr net.Addr, err error) { c.l.Lock() defer c.l.Unlock() if c.address != nil { return c.address, nil } // If one of cmd or reattach isn't set, then it is an error. We wrap // this in a {} for scoping reasons, and hopeful that the escape // analysis will pop the stock here. { cmdSet := c.config.Cmd != nil attachSet := c.config.Reattach != nil secureSet := c.config.SecureConfig != nil if cmdSet == attachSet { return nil, fmt.Errorf("Only one of Cmd or Reattach must be set") } if secureSet && attachSet { return nil, ErrSecureConfigAndReattach } } // Create the logging channel for when we kill c.doneLogging = make(chan struct{}) // Create a context for when we kill var ctxCancel context.CancelFunc c.doneCtx, ctxCancel = context.WithCancel(context.Background()) if c.config.Reattach != nil { // Verify the process still exists. If not, then it is an error p, err := os.FindProcess(c.config.Reattach.Pid) if err != nil { return nil, err } // Attempt to connect to the addr since on Unix systems FindProcess // doesn't actually return an error if it can't find the process. conn, err := net.Dial( c.config.Reattach.Addr.Network(), c.config.Reattach.Addr.String()) if err != nil { p.Kill() return nil, ErrProcessNotFound } conn.Close() // Goroutine to mark exit status go func(pid int) { // Wait for the process to die pidWait(pid) // Log so we can see it c.logger.Debug("reattached plugin process exited") // Mark it c.l.Lock() defer c.l.Unlock() c.exited = true // Close the logging channel since that doesn't work on reattach close(c.doneLogging) // Cancel the context ctxCancel() }(p.Pid) // Set the address and process c.address = c.config.Reattach.Addr c.process = p c.protocol = c.config.Reattach.Protocol if c.protocol == "" { // Default the protocol to net/rpc for backwards compatibility c.protocol = ProtocolNetRPC } return c.address, nil } env := []string{ fmt.Sprintf("%s=%s", c.config.MagicCookieKey, c.config.MagicCookieValue), fmt.Sprintf("PLUGIN_MIN_PORT=%d", c.config.MinPort), fmt.Sprintf("PLUGIN_MAX_PORT=%d", c.config.MaxPort), } stdout_r, stdout_w := io.Pipe() stderr_r, stderr_w := io.Pipe() cmd := c.config.Cmd cmd.Env = append(cmd.Env, os.Environ()...) cmd.Env = append(cmd.Env, env...) cmd.Stdin = os.Stdin cmd.Stderr = stderr_w cmd.Stdout = stdout_w if c.config.SecureConfig != nil { if ok, err := c.config.SecureConfig.Check(cmd.Path); err != nil { return nil, fmt.Errorf("error verifying checksum: %s", err) } else if !ok { return nil, ErrChecksumsDoNotMatch } } c.logger.Debug("starting plugin", "path", cmd.Path, "args", cmd.Args) err = cmd.Start() if err != nil { return } // Set the process c.process = cmd.Process // Make sure the command is properly cleaned up if there is an error defer func() { r := recover() if err != nil || r != nil { cmd.Process.Kill() } if r != nil { panic(r) } }() // Start goroutine to wait for process to exit exitCh := make(chan struct{}) go func() { // Make sure we close the write end of our stderr/stdout so // that the readers send EOF properly. defer stderr_w.Close() defer stdout_w.Close() // Wait for the command to end. cmd.Wait() // Log and make sure to flush the logs write away c.logger.Debug("plugin process exited", "path", cmd.Path) os.Stderr.Sync() // Mark that we exited close(exitCh) // Cancel the context, marking that we exited ctxCancel() // Set that we exited, which takes a lock c.l.Lock() defer c.l.Unlock() c.exited = true }() // Start goroutine that logs the stderr go c.logStderr(stderr_r) // Start a goroutine that is going to be reading the lines // out of stdout linesCh := make(chan []byte) go func() { defer close(linesCh) buf := bufio.NewReader(stdout_r) for { line, err := buf.ReadBytes('\n') if line != nil { linesCh <- line } if err == io.EOF { return } } }() // Make sure after we exit we read the lines from stdout forever // so they don't block since it is an io.Pipe defer func() { go func() { for _ = range linesCh { } }() }() // Some channels for the next step timeout := time.After(c.config.StartTimeout) // Start looking for the address c.logger.Debug("waiting for RPC address", "path", cmd.Path) select { case <-timeout: err = errors.New("timeout while waiting for plugin to start") case <-exitCh: err = errors.New("plugin exited before we could connect") case lineBytes := <-linesCh: // Trim the line and split by "|" in order to get the parts of // the output. line := strings.TrimSpace(string(lineBytes)) parts := strings.SplitN(line, "|", 6) if len(parts) < 4 { err = fmt.Errorf( "Unrecognized remote plugin message: %s\n\n"+ "This usually means that the plugin is either invalid or simply\n"+ "needs to be recompiled to support the latest protocol.", line) return } // Check the core protocol. Wrapped in a {} for scoping. { var coreProtocol int64 coreProtocol, err = strconv.ParseInt(parts[0], 10, 0) if err != nil { err = fmt.Errorf("Error parsing core protocol version: %s", err) return } if int(coreProtocol) != CoreProtocolVersion { err = fmt.Errorf("Incompatible core API version with plugin. "+ "Plugin version: %s, Core version: %d\n\n"+ "To fix this, the plugin usually only needs to be recompiled.\n"+ "Please report this to the plugin author.", parts[0], CoreProtocolVersion) return } } // Parse the protocol version var protocol int64 protocol, err = strconv.ParseInt(parts[1], 10, 0) if err != nil { err = fmt.Errorf("Error parsing protocol version: %s", err) return } // Test the API version if uint(protocol) != c.config.ProtocolVersion { err = fmt.Errorf("Incompatible API version with plugin. "+ "Plugin version: %s, Core version: %d", parts[1], c.config.ProtocolVersion) return } switch parts[2] { case "tcp": addr, err = net.ResolveTCPAddr("tcp", parts[3]) case "unix": addr, err = net.ResolveUnixAddr("unix", parts[3]) default: err = fmt.Errorf("Unknown address type: %s", parts[3]) } // If we have a server type, then record that. We default to net/rpc // for backwards compatibility. c.protocol = ProtocolNetRPC if len(parts) >= 5 { c.protocol = Protocol(parts[4]) } found := false for _, p := range c.config.AllowedProtocols { if p == c.protocol { found = true break } } if !found { err = fmt.Errorf("Unsupported plugin protocol %q. Supported: %v", c.protocol, c.config.AllowedProtocols) return } } c.address = addr return } // ReattachConfig returns the information that must be provided to NewClient // to reattach to the plugin process that this client started. This is // useful for plugins that detach from their parent process. // // If this returns nil then the process hasn't been started yet. Please // call Start or Client before calling this. func (c *Client) ReattachConfig() *ReattachConfig { c.l.Lock() defer c.l.Unlock() if c.address == nil { return nil } if c.config.Cmd != nil && c.config.Cmd.Process == nil { return nil } // If we connected via reattach, just return the information as-is if c.config.Reattach != nil { return c.config.Reattach } return &ReattachConfig{ Protocol: c.protocol, Addr: c.address, Pid: c.config.Cmd.Process.Pid, } } // Protocol returns the protocol of server on the remote end. This will // start the plugin process if it isn't already started. Errors from // starting the plugin are surpressed and ProtocolInvalid is returned. It // is recommended you call Start explicitly before calling Protocol to ensure // no errors occur. func (c *Client) Protocol() Protocol { _, err := c.Start() if err != nil { return ProtocolInvalid } return c.protocol } func netAddrDialer(addr net.Addr) func(string, time.Duration) (net.Conn, error) { return func(_ string, _ time.Duration) (net.Conn, error) { // Connect to the client conn, err := net.Dial(addr.Network(), addr.String()) if err != nil { return nil, err } if tcpConn, ok := conn.(*net.TCPConn); ok { // Make sure to set keep alive so that the connection doesn't die tcpConn.SetKeepAlive(true) } return conn, nil } } // dialer is compatible with grpc.WithDialer and creates the connection // to the plugin. func (c *Client) dialer(_ string, timeout time.Duration) (net.Conn, error) { conn, err := netAddrDialer(c.address)("", timeout) if err != nil { return nil, err } // If we have a TLS config we wrap our connection. We only do this // for net/rpc since gRPC uses its own mechanism for TLS. if c.protocol == ProtocolNetRPC && c.config.TLSConfig != nil { conn = tls.Client(conn, c.config.TLSConfig) } return conn, nil } func (c *Client) logStderr(r io.Reader) { bufR := bufio.NewReader(r) l := c.logger.Named(filepath.Base(c.config.Cmd.Path)) for { line, err := bufR.ReadString('\n') if line != "" { c.config.Stderr.Write([]byte(line)) line = strings.TrimRightFunc(line, unicode.IsSpace) entry, err := parseJSON(line) // If output is not JSON format, print directly to Debug if err != nil { l.Debug(line) } else { out := flattenKVPairs(entry.KVPairs) out = append(out, "timestamp", entry.Timestamp.Format(hclog.TimeFormat)) switch hclog.LevelFromString(entry.Level) { case hclog.Trace: l.Trace(entry.Message, out...) case hclog.Debug: l.Debug(entry.Message, out...) case hclog.Info: l.Info(entry.Message, out...) case hclog.Warn: l.Warn(entry.Message, out...) case hclog.Error: l.Error(entry.Message, out...) } } } if err == io.EOF { break } } // Flag that we've completed logging for others close(c.doneLogging) }