file actions: add bucket and endpoint to notifications
The HTTP notifications are now invoked as POST and the notification is a JSON inside the POST body. This is a backward incompatible change but this way the actions can be extended more easily, sorry for the trouble Fixes #101
This commit is contained in:
parent
e22d377203
commit
4759254e10
7 changed files with 149 additions and 72 deletions
|
@ -7,7 +7,7 @@ For each account, the following properties can be configured:
|
|||
- `public_keys` array of public keys. At least one public key or the password is mandatory.
|
||||
- `status` 1 means "active", 0 "inactive". An inactive account cannot login.
|
||||
- `expiration_date` expiration date as unix timestamp in milliseconds. An expired account cannot login. 0 means no expiration.
|
||||
- `home_dir` the user cannot upload or download files outside this directory. Must be an absolute path.
|
||||
- `home_dir` the user cannot upload or download files outside this directory. Must be an absolute path. A local home directory is required for Cloud Storage Backends too: in this case it will store temporary files.
|
||||
- `virtual_folders` list of mappings between virtual SFTP/SCP paths and local filesystem paths outside the user home directory. The specified paths must be absolute and the virtual path cannot be "/", it must be a sub directory. The parent directory for the specified virtual path must exist. SFTPGo will try to automatically create any missing parent directory for the configured virtual folders at user login
|
||||
- `uid`, `gid`. If SFTPGo runs as root system user then the created files and directories will be assigned to this system uid/gid. Ignored on windows or if SFTPGo runs as non root user: in this case files and directories for all SFTP users will be owned by the system user that runs SFTPGo.
|
||||
- `max_sessions` maximum concurrent sessions. 0 means unlimited.
|
||||
|
|
|
@ -9,31 +9,36 @@ The `command`, if defined, is invoked with the following arguments:
|
|||
- `action`, string, possible values are: `download`, `upload`, `delete`, `rename`, `ssh_cmd`
|
||||
- `username`
|
||||
- `path` is the full filesystem path, can be empty for some ssh commands
|
||||
- `target_path`, non empty for `rename` action
|
||||
- `ssh_cmd`, non empty for `ssh_cmd` action
|
||||
- `target_path`, non-empty for `rename` action
|
||||
- `ssh_cmd`, non-empty for `ssh_cmd` action
|
||||
|
||||
The `command` can also read the following environment variables:
|
||||
|
||||
- `SFTPGO_ACTION`
|
||||
- `SFTPGO_ACTION_USERNAME`
|
||||
- `SFTPGO_ACTION_PATH`
|
||||
- `SFTPGO_ACTION_TARGET`, non empty for `rename` `SFTPGO_ACTION`
|
||||
- `SFTPGO_ACTION_SSH_CMD`, non empty for `ssh_cmd` `SFTPGO_ACTION`
|
||||
- `SFTPGO_ACTION_FILE_SIZE`, non empty for `upload`, `download` and `delete` `SFTPGO_ACTION`
|
||||
- `SFTPGO_ACTION_LOCAL_FILE`, `true` if the affected file is stored on the local filesystem, otherwise `false`
|
||||
- `SFTPGO_ACTION_TARGET`, non-empty for `rename` `SFTPGO_ACTION`
|
||||
- `SFTPGO_ACTION_SSH_CMD`, non-empty for `ssh_cmd` `SFTPGO_ACTION`
|
||||
- `SFTPGO_ACTION_FILE_SIZE`, non-empty for `upload`, `download` and `delete` `SFTPGO_ACTION`
|
||||
- `SFTPGO_ACTION_FS_PROVIDER`, `0` for local filesystem, `1` for S3 backend, `2` for Google Cloud Storage (GCS) backend
|
||||
- `SFTPGO_ACTION_BUCKET`, non-empty for S3 and GCS backends
|
||||
- `SFTPGO_ACTION_ENDPOINT`, non-empty for S3 backend if configured
|
||||
|
||||
Previous global environment variables aren't cleared when the script is called.
|
||||
The `command` must finish within 30 seconds.
|
||||
|
||||
The `http_notification_url`, if defined, will contain the following, percent encoded, query string parameters:
|
||||
The `http_notification_url`, if defined, will be invoked as HTTP POST. The request body will contain a JSON serialized struct with the following fields:
|
||||
|
||||
- `action`
|
||||
- `username`
|
||||
- `path`
|
||||
- `local_file`, `true` if the affected file is stored on the local filesystem, otherwise `false`
|
||||
- `target_path`, added for `rename` action
|
||||
- `ssh_cmd`, added for `ssh_cmd` action
|
||||
- `file_size`, added for `upload`, `download`, `delete` actions
|
||||
- `target_path`, not null for `rename` action
|
||||
- `ssh_cmd`, not null for `ssh_cmd` action
|
||||
- `file_size`, not null for `upload`, `download`, `delete` actions
|
||||
- `fs_provider`, `0` for local filesystem, `1` for S3 backend, `2` for Google Cloud Storage (GCS) backend
|
||||
- `bucket`, not null for S3 and GCS backends
|
||||
- `endpoint`, not null for S3 backend if configured
|
||||
|
||||
|
||||
The HTTP request is executed with a 15-second timeout.
|
||||
|
||||
|
@ -73,6 +78,6 @@ The `command` can also read the following environment variables:
|
|||
Previous global environment variables aren't cleared when the script is called.
|
||||
The `command` must finish within 15 seconds.
|
||||
|
||||
The `http_notification_url`, if defined, will be called invoked as http POST. The action is added to the query string, for example `<http_notification_url>?action=update`, and the user is sent serialized as JSON inside the POST body with sensitive fields removed.
|
||||
The `http_notification_url`, if defined, will be invoked as HTTP POST. The action is added to the query string, for example `<http_notification_url>?action=update`, and the user is sent serialized as JSON inside the POST body with sensitive fields removed.
|
||||
|
||||
The HTTP request is executed with a 15-second timeout.
|
||||
|
|
|
@ -335,7 +335,7 @@ func (c Connection) handleSFTPRename(sourcePath string, targetPath string, reque
|
|||
return vfs.GetSFTPError(c.fs, err)
|
||||
}
|
||||
logger.CommandLog(renameLogSender, sourcePath, targetPath, c.User.Username, "", c.ID, c.protocol, -1, -1, "", "", "")
|
||||
go executeAction(operationRename, c.User.Username, sourcePath, targetPath, "", 0, vfs.IsLocalOsFs(c.fs))
|
||||
go executeAction(newActionNotification(c.User, operationRename, sourcePath, targetPath, "", 0))
|
||||
return nil
|
||||
}
|
||||
|
||||
|
@ -443,7 +443,7 @@ func (c Connection) handleSFTPRemove(filePath string, request *sftp.Request) err
|
|||
if fi.Mode()&os.ModeSymlink != os.ModeSymlink {
|
||||
dataprovider.UpdateUserQuota(dataProvider, c.User, -1, -size, false)
|
||||
}
|
||||
go executeAction(operationDelete, c.User.Username, filePath, "", "", fi.Size(), vfs.IsLocalOsFs(c.fs))
|
||||
go executeAction(newActionNotification(c.User, operationDelete, filePath, "", "", fi.Size()))
|
||||
|
||||
return sftp.ErrSSHFxOk
|
||||
}
|
||||
|
|
|
@ -119,6 +119,46 @@ func newMockOsFs(err, statErr error, atomicUpload bool, connectionID, rootDir st
|
|||
}
|
||||
}
|
||||
|
||||
func TestNewActionNotification(t *testing.T) {
|
||||
user := dataprovider.User{
|
||||
Username: "username",
|
||||
}
|
||||
user.FsConfig.Provider = 0
|
||||
user.FsConfig.S3Config = vfs.S3FsConfig{
|
||||
Bucket: "s3bucket",
|
||||
Endpoint: "endpoint",
|
||||
}
|
||||
user.FsConfig.GCSConfig = vfs.GCSFsConfig{
|
||||
Bucket: "gcsbucket",
|
||||
}
|
||||
a := newActionNotification(user, operationDownload, "path", "target", "", 123)
|
||||
if a.Username != "username" {
|
||||
t.Errorf("unexpected username")
|
||||
}
|
||||
if len(a.Bucket) > 0 {
|
||||
t.Errorf("unexpected bucket")
|
||||
}
|
||||
if len(a.Endpoint) > 0 {
|
||||
t.Errorf("unexpected endpoint")
|
||||
}
|
||||
user.FsConfig.Provider = 1
|
||||
a = newActionNotification(user, operationDownload, "path", "target", "", 123)
|
||||
if a.Bucket != "s3bucket" {
|
||||
t.Errorf("unexpected s3 bucket")
|
||||
}
|
||||
if a.Endpoint != "endpoint" {
|
||||
t.Errorf("unexpected endpoint")
|
||||
}
|
||||
user.FsConfig.Provider = 2
|
||||
a = newActionNotification(user, operationDownload, "path", "target", "", 123)
|
||||
if a.Bucket != "gcsbucket" {
|
||||
t.Errorf("unexpected gcs bucket")
|
||||
}
|
||||
if len(a.Endpoint) > 0 {
|
||||
t.Errorf("unexpected endpoint")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWrongActions(t *testing.T) {
|
||||
actionsCopy := actions
|
||||
badCommand := "/bad/command"
|
||||
|
@ -130,17 +170,20 @@ func TestWrongActions(t *testing.T) {
|
|||
Command: badCommand,
|
||||
HTTPNotificationURL: "",
|
||||
}
|
||||
err := executeAction(operationDownload, "username", "path", "", "", 0, true)
|
||||
user := dataprovider.User{
|
||||
Username: "username",
|
||||
}
|
||||
err := executeAction(newActionNotification(user, operationDownload, "path", "", "", 0))
|
||||
if err == nil {
|
||||
t.Errorf("action with bad command must fail")
|
||||
}
|
||||
err = executeAction(operationDelete, "username", "path", "", "", 0, true)
|
||||
err = executeAction(newActionNotification(user, operationDelete, "path", "", "", 0))
|
||||
if err != nil {
|
||||
t.Errorf("action not configured must silently fail")
|
||||
}
|
||||
actions.Command = ""
|
||||
actions.HTTPNotificationURL = "http://foo\x7f.com/"
|
||||
err = executeAction(operationDownload, "username", "path", "", "", 0, true)
|
||||
err = executeAction(newActionNotification(user, operationDownload, "path", "", "", 0))
|
||||
if err == nil {
|
||||
t.Errorf("action with bad url must fail")
|
||||
}
|
||||
|
@ -154,7 +197,10 @@ func TestActionHTTP(t *testing.T) {
|
|||
Command: "",
|
||||
HTTPNotificationURL: "http://127.0.0.1:8080/",
|
||||
}
|
||||
err := executeAction(operationDownload, "username", "path", "", "", 0, true)
|
||||
user := dataprovider.User{
|
||||
Username: "username",
|
||||
}
|
||||
err := executeAction(newActionNotification(user, operationDownload, "path", "", "", 0))
|
||||
if err != nil {
|
||||
t.Errorf("unexpected error: %v", err)
|
||||
}
|
||||
|
|
126
sftpd/sftpd.go
126
sftpd/sftpd.go
|
@ -4,7 +4,9 @@
|
|||
package sftpd
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
|
@ -127,6 +129,57 @@ type sshSubsystemExecMsg struct {
|
|||
Command string
|
||||
}
|
||||
|
||||
type actionNotification struct {
|
||||
Action string `json:"action"`
|
||||
Username string `json:"username"`
|
||||
Path string `json:"path"`
|
||||
TargetPath string `json:"target_path,omitempty"`
|
||||
SSHCmd string `json:"ssh_cmd,omitempty"`
|
||||
FileSize int64 `json:"file_size,omitempty"`
|
||||
FsProvider int `json:"fs_provider"`
|
||||
Bucket string `json:"bucket,omitempty"`
|
||||
Endpoint string `json:"endpoint,omitempty"`
|
||||
}
|
||||
|
||||
func newActionNotification(user dataprovider.User, operation, filePath, target, sshCmd string, fileSize int64) actionNotification {
|
||||
bucket := ""
|
||||
endpoint := ""
|
||||
if user.FsConfig.Provider == 1 {
|
||||
bucket = user.FsConfig.S3Config.Bucket
|
||||
endpoint = user.FsConfig.S3Config.Endpoint
|
||||
} else if user.FsConfig.Provider == 2 {
|
||||
bucket = user.FsConfig.GCSConfig.Bucket
|
||||
}
|
||||
return actionNotification{
|
||||
Action: operation,
|
||||
Username: user.Username,
|
||||
Path: filePath,
|
||||
TargetPath: target,
|
||||
SSHCmd: sshCmd,
|
||||
FileSize: fileSize,
|
||||
FsProvider: user.FsConfig.Provider,
|
||||
Bucket: bucket,
|
||||
Endpoint: endpoint,
|
||||
}
|
||||
}
|
||||
|
||||
func (a *actionNotification) AsJSON() []byte {
|
||||
res, _ := json.Marshal(a)
|
||||
return res
|
||||
}
|
||||
|
||||
func (a *actionNotification) AsEnvVars() []string {
|
||||
return []string{fmt.Sprintf("SFTPGO_ACTION=%v", a.Action),
|
||||
fmt.Sprintf("SFTPGO_ACTION_USERNAME=%v", a.Username),
|
||||
fmt.Sprintf("SFTPGO_ACTION_PATH=%v", a.Path),
|
||||
fmt.Sprintf("SFTPGO_ACTION_TARGET=%v", a.TargetPath),
|
||||
fmt.Sprintf("SFTPGO_ACTION_SSH_CMD=%v", a.SSHCmd),
|
||||
fmt.Sprintf("SFTPGO_ACTION_FILE_SIZE=%v", a.FileSize),
|
||||
fmt.Sprintf("SFTPGO_ACTION_FS_PROVIDER=%v", a.FsProvider),
|
||||
fmt.Sprintf("SFTPGO_ACTION_BUCKET=%v", a.Bucket),
|
||||
fmt.Sprintf("SFTPGO_ACTION_ENDPOINT=%v", a.Endpoint)}
|
||||
}
|
||||
|
||||
func init() {
|
||||
openConnections = make(map[string]Connection)
|
||||
idleConnectionTicker = time.NewTicker(5 * time.Minute)
|
||||
|
@ -415,80 +468,53 @@ func isAtomicUploadEnabled() bool {
|
|||
return uploadMode == uploadModeAtomic || uploadMode == uploadModeAtomicWithResume
|
||||
}
|
||||
|
||||
func executeNotificationCommand(operation, username, path, target, sshCmd, fileSize, isLocalFile string) error {
|
||||
func executeNotificationCommand(a actionNotification) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
cmd := exec.CommandContext(ctx, actions.Command, operation, username, path, target, sshCmd)
|
||||
cmd.Env = append(os.Environ(),
|
||||
fmt.Sprintf("SFTPGO_ACTION=%v", operation),
|
||||
fmt.Sprintf("SFTPGO_ACTION_USERNAME=%v", username),
|
||||
fmt.Sprintf("SFTPGO_ACTION_PATH=%v", path),
|
||||
fmt.Sprintf("SFTPGO_ACTION_TARGET=%v", target),
|
||||
fmt.Sprintf("SFTPGO_ACTION_SSH_CMD=%v", sshCmd),
|
||||
fmt.Sprintf("SFTPGO_ACTION_FILE_SIZE=%v", fileSize),
|
||||
fmt.Sprintf("SFTPGO_ACTION_LOCAL_FILE=%v", isLocalFile),
|
||||
)
|
||||
cmd := exec.CommandContext(ctx, actions.Command, a.Action, a.Username, a.Path, a.TargetPath, a.SSHCmd)
|
||||
cmd.Env = append(os.Environ(), a.AsEnvVars()...)
|
||||
startTime := time.Now()
|
||||
err := cmd.Run()
|
||||
logger.Debug(logSender, "", "executed command %#v with arguments: %#v, %#v, %#v, %#v, %#v, elapsed: %v, error: %v",
|
||||
actions.Command, operation, username, path, target, sshCmd, time.Since(startTime), err)
|
||||
actions.Command, a.Action, a.Username, a.Path, a.TargetPath, a.SSHCmd, time.Since(startTime), err)
|
||||
return err
|
||||
}
|
||||
|
||||
// executed in a goroutine
|
||||
func executeAction(operation, username, path, target, sshCmd string, fileSize int64, isLocalFile bool) error {
|
||||
if !utils.IsStringInSlice(operation, actions.ExecuteOn) {
|
||||
func executeAction(a actionNotification) error {
|
||||
if !utils.IsStringInSlice(a.Action, actions.ExecuteOn) {
|
||||
return nil
|
||||
}
|
||||
var err error
|
||||
size := ""
|
||||
if fileSize > 0 {
|
||||
size = fmt.Sprintf("%v", fileSize)
|
||||
}
|
||||
if len(actions.Command) > 0 && filepath.IsAbs(actions.Command) {
|
||||
// we are in a goroutine but if we have to send an HTTP notification we don't want to wait for the
|
||||
// end of the command
|
||||
if len(actions.HTTPNotificationURL) > 0 {
|
||||
go executeNotificationCommand(operation, username, path, target, sshCmd, size, fmt.Sprintf("%t", isLocalFile))
|
||||
go executeNotificationCommand(a)
|
||||
} else {
|
||||
err = executeNotificationCommand(operation, username, path, target, sshCmd, size, fmt.Sprintf("%t", isLocalFile))
|
||||
err = executeNotificationCommand(a)
|
||||
}
|
||||
}
|
||||
if len(actions.HTTPNotificationURL) > 0 {
|
||||
var url *url.URL
|
||||
url, err = url.Parse(actions.HTTPNotificationURL)
|
||||
if err == nil {
|
||||
q := url.Query()
|
||||
q.Add("action", operation)
|
||||
q.Add("username", username)
|
||||
q.Add("path", path)
|
||||
if len(target) > 0 {
|
||||
q.Add("target_path", target)
|
||||
}
|
||||
if len(sshCmd) > 0 {
|
||||
q.Add("ssh_cmd", sshCmd)
|
||||
}
|
||||
if len(size) > 0 {
|
||||
q.Add("file_size", size)
|
||||
}
|
||||
q.Add("local_file", fmt.Sprintf("%t", isLocalFile))
|
||||
url.RawQuery = q.Encode()
|
||||
startTime := time.Now()
|
||||
httpClient := &http.Client{
|
||||
Timeout: 15 * time.Second,
|
||||
}
|
||||
resp, err := httpClient.Get(url.String())
|
||||
respCode := 0
|
||||
if err == nil {
|
||||
respCode = resp.StatusCode
|
||||
resp.Body.Close()
|
||||
}
|
||||
logger.Debug(logSender, "", "notified operation %#v to URL: %v status code: %v, elapsed: %v err: %v",
|
||||
operation, url.String(), respCode, time.Since(startTime), err)
|
||||
} else {
|
||||
if err != nil {
|
||||
logger.Warn(logSender, "", "Invalid http_notification_url %#v for operation %#v: %v", actions.HTTPNotificationURL,
|
||||
operation, err)
|
||||
a.Action, err)
|
||||
return err
|
||||
}
|
||||
startTime := time.Now()
|
||||
httpClient := &http.Client{
|
||||
Timeout: 15 * time.Second,
|
||||
}
|
||||
resp, err := httpClient.Post(url.String(), "application/json", bytes.NewBuffer(a.AsJSON()))
|
||||
respCode := 0
|
||||
if err == nil {
|
||||
respCode = resp.StatusCode
|
||||
resp.Body.Close()
|
||||
}
|
||||
logger.Debug(logSender, "", "notified operation %#v to URL: %v status code: %v, elapsed: %v err: %v",
|
||||
a.Action, url.String(), respCode, time.Since(startTime), err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
|
|
@ -454,7 +454,7 @@ func (c *sshCommand) sendExitStatus(err error) {
|
|||
realPath = p
|
||||
}
|
||||
}
|
||||
go executeAction(operationSSHCmd, c.connection.User.Username, realPath, "", c.command, 0, vfs.IsLocalOsFs(c.connection.fs))
|
||||
go executeAction(newActionNotification(c.connection.User, operationSSHCmd, realPath, "", c.command, 0))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -151,10 +151,10 @@ func (t *Transfer) Close() error {
|
|||
elapsed := time.Since(t.start).Nanoseconds() / 1000000
|
||||
if t.transferType == transferDownload {
|
||||
logger.TransferLog(downloadLogSender, t.path, elapsed, t.bytesSent, t.user.Username, t.connectionID, t.protocol)
|
||||
go executeAction(operationDownload, t.user.Username, t.path, "", "", t.bytesSent, (t.file != nil))
|
||||
go executeAction(newActionNotification(t.user, operationDownload, t.path, "", "", t.bytesSent))
|
||||
} else {
|
||||
logger.TransferLog(uploadLogSender, t.path, elapsed, t.bytesReceived, t.user.Username, t.connectionID, t.protocol)
|
||||
go executeAction(operationUpload, t.user.Username, t.path, "", "", t.bytesReceived+t.minWriteOffset, (t.file != nil))
|
||||
go executeAction(newActionNotification(t.user, operationUpload, t.path, "", "", t.bytesReceived+t.minWriteOffset))
|
||||
}
|
||||
} else {
|
||||
logger.Warn(logSender, t.connectionID, "transfer error: %v, path: %#v", t.transferError, t.path)
|
||||
|
|
Loading…
Reference in a new issue