Merge pull request #1588 from fatedier/dev

bump version to v0.31.0

README.md

@@ -37,6 +37,8 @@ frp also has a P2P connect mode.
     * [Get proxy status from client](#get-proxy-status-from-client)
     * [Only allowing certain ports on the server](#only-allowing-certain-ports-on-the-server)
     * [Port Reuse](#port-reuse)
+    * [Bandwidth Limit](#bandwidth-limit)
+        * [For Each Proxy](#for-each-proxy)
     * [TCP Stream Multiplexing](#tcp-stream-multiplexing)
     * [Support KCP Protocol](#support-kcp-protocol)
     * [Connection Pooling](#connection-pooling)
@@ -52,7 +54,8 @@ frp also has a P2P connect mode.
     * [URL routing](#url-routing)
     * [Connecting to frps via HTTP PROXY](#connecting-to-frps-via-http-proxy)
     * [Range ports mapping](#range-ports-mapping)
-    * [Plugins](#plugins)
+    * [Client Plugins](#client-plugins)
+    * [Server Manage Plugins](#server-manage-plugins)
 * [Development Plan](#development-plan)
 * [Contributing](#contributing)
 * [Donation](#donation)
@@ -495,6 +498,21 @@ allow_ports = 2000-3000,3001,3003,4000-50000
 
 We would like to try to allow multiple proxies bind a same remote port with different protocols in the future.
 
+### Bandwidth Limit
+
+#### For Each Proxy
+
+```ini
+# frpc.ini
+[ssh]
+type = tcp
+local_port = 22
+remote_port = 6000
+bandwidth_limit = 1MB
+```
+
+Set `bandwidth_limit` in each proxy's configure to enable this feature. Supported units are `MB` and `KB`.
+
 ### TCP Stream Multiplexing
 
 frp supports tcp stream multiplexing since v0.10.0 like HTTP2 Multiplexing, in which case all logic connections to the same frpc are multiplexed into the same TCP connection.
@@ -789,7 +807,7 @@ remote_port = 6000-6006,6007
 
 frpc will generate 8 proxies like `test_tcp_0`, `test_tcp_1`, ..., `test_tcp_7`.
 
-### Plugins
+### Client Plugins
 
 frpc only forwards requests to local TCP or UDP ports by default.
 
@@ -811,6 +829,10 @@ plugin_http_passwd = abc
 
 `plugin_http_user` and `plugin_http_passwd` are configuration parameters used in `http_proxy` plugin.
 
+### Server Manage Plugins
+
+Read the [document](/doc/server_plugin.md).
+
 ## Development Plan
 
 * Log HTTP request information in frps.

README_zh.md

@@ -33,6 +33,8 @@ frp 是一个可用于内网穿透的高性能的反向代理应用，支持 tcp
     * [客户端查看代理状态](#客户端查看代理状态)
     * [端口白名单](#端口白名单)
     * [端口复用](#端口复用)
+    * [限速](#限速)
+        * [代理限速](#代理限速)
     * [TCP 多路复用](#tcp-多路复用)
     * [底层通信可选 kcp 协议](#底层通信可选-kcp-协议)
     * [连接池](#连接池)
@@ -48,7 +50,8 @@ frp 是一个可用于内网穿透的高性能的反向代理应用，支持 tcp
     * [URL 路由](#url-路由)
     * [通过代理连接 frps](#通过代理连接-frps)
     * [范围端口映射](#范围端口映射)
-    * [插件](#插件)
+    * [客户端插件](#客户端插件)
+    * [服务端管理插件](#服务端管理插件)
 * [开发计划](#开发计划)
 * [为 frp 做贡献](#为-frp-做贡献)
 * [捐助](#捐助)
@@ -531,6 +534,23 @@ allow_ports = 2000-3000,3001,3003,4000-50000
 
 后续会尝试允许多个 proxy 绑定同一个远端端口的不同协议。
 
+### 限速
+
+#### 代理限速
+
+目前支持在客户端的代理配置中设置代理级别的限速，限制单个 proxy 可以占用的带宽。
+
+```ini
+# frpc.ini
+[ssh]
+type = tcp
+local_port = 22
+remote_port = 6000
+bandwith_limit = 1MB
+```
+
+在代理配置中增加 `bandwith_limit` 字段启用此功能，目前仅支持 `MB` 和 `KB` 单位。
+
 ### TCP 多路复用
 
 从 v0.10.0 版本开始，客户端和服务器端之间的连接支持多路复用，不再需要为每一个用户请求创建一个连接，使连接建立的延迟降低，并且避免了大量文件描述符的占用，使 frp 可以承载更高的并发数。
@@ -839,11 +859,11 @@ remote_port = 6000-6006,6007
 
 实际连接成功后会创建 8 个 proxy，命名为 `test_tcp_0, test_tcp_1 ... test_tcp_7`。
 
-### 插件
+### 客户端插件
 
 默认情况下，frpc 只会转发请求到本地 tcp 或 udp 端口。
 
-插件模式是为了在客户端提供更加丰富的功能，目前内置的插件有 `unix_domain_socket`、`http_proxy`、`socks5`、`static_file`。具体使用方式请查看[使用示例](#使用示例)。
+客户端插件模式是为了在客户端提供更加丰富的功能，目前内置的插件有 `unix_domain_socket`、`http_proxy`、`socks5`、`static_file`。具体使用方式请查看[使用示例](#使用示例)。
 
 通过 `plugin` 指定需要使用的插件，插件的配置参数都以 `plugin_` 开头。使用插件后 `local_ip` 和 `local_port` 不再需要配置。
 
@@ -861,6 +881,10 @@ plugin_http_passwd = abc
 
 `plugin_http_user` 和 `plugin_http_passwd` 即为 `http_proxy` 插件可选的配置参数。
 
+### 服务端管理插件
+
+[使用说明](/doc/server_plugin_zh.md)
+
 ## 开发计划
 
 计划在后续版本中加入的功能与优化，排名不分先后，如果有其他功能建议欢迎在 [issues](https://github.com/fatedier/frp/issues) 中反馈。

assets/assets.go

@@ -55,6 +55,7 @@ func ReadFile(file string) (content string, err error) {
 		if err != nil {
 			return content, err
 		}
+		defer file.Close()
 		buf, err := ioutil.ReadAll(file)
 		if err != nil {
 			return content, err
@@ -65,6 +66,7 @@ func ReadFile(file string) (content string, err error) {
 		if err != nil {
 			return content, err
 		}
+		defer file.Close()
 		buf, err := ioutil.ReadAll(file)
 		if err != nil {
 			return content, err

client/proxy/proxy.go

@@ -28,8 +28,9 @@ import (
 
 	"github.com/fatedier/frp/models/config"
 	"github.com/fatedier/frp/models/msg"
-	"github.com/fatedier/frp/models/plugin"
+	plugin "github.com/fatedier/frp/models/plugin/client"
 	"github.com/fatedier/frp/models/proto/udp"
+	"github.com/fatedier/frp/utils/limit"
 	frpNet "github.com/fatedier/frp/utils/net"
 	"github.com/fatedier/frp/utils/xlog"
 
@@ -38,6 +39,7 @@ import (
 	"github.com/fatedier/golib/pool"
 	fmux "github.com/hashicorp/yamux"
 	pp "github.com/pires/go-proxyproto"
+	"golang.org/x/time/rate"
 )
 
 // Proxy defines how to handle work connections for different proxy type.
@@ -51,9 +53,16 @@ type Proxy interface {
 }
 
 func NewProxy(ctx context.Context, pxyConf config.ProxyConf, clientCfg config.ClientCommonConf, serverUDPPort int) (pxy Proxy) {
+	var limiter *rate.Limiter
+	limitBytes := pxyConf.GetBaseInfo().BandwidthLimit.Bytes()
+	if limitBytes > 0 {
+		limiter = rate.NewLimiter(rate.Limit(float64(limitBytes)), int(limitBytes))
+	}
+
 	baseProxy := BaseProxy{
 		clientCfg:     clientCfg,
 		serverUDPPort: serverUDPPort,
+		limiter:       limiter,
 		xl:            xlog.FromContextSafe(ctx),
 		ctx:           ctx,
 	}
@@ -96,6 +105,7 @@ type BaseProxy struct {
 	closed        bool
 	clientCfg     config.ClientCommonConf
 	serverUDPPort int
+	limiter       *rate.Limiter
 
 	mu  sync.RWMutex
 	xl  *xlog.Logger
@@ -127,8 +137,8 @@ func (pxy *TcpProxy) Close() {
 }
 
 func (pxy *TcpProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
-	HandleTcpWorkConnection(pxy.ctx, &pxy.cfg.LocalSvrConf, pxy.proxyPlugin, &pxy.cfg.BaseProxyConf, conn,
-		[]byte(pxy.clientCfg.Token), m)
+	HandleTcpWorkConnection(pxy.ctx, &pxy.cfg.LocalSvrConf, pxy.proxyPlugin, &pxy.cfg.BaseProxyConf, pxy.limiter,
+		conn, []byte(pxy.clientCfg.Token), m)
 }
 
 // HTTP
@@ -156,8 +166,8 @@ func (pxy *HttpProxy) Close() {
 }
 
 func (pxy *HttpProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
-	HandleTcpWorkConnection(pxy.ctx, &pxy.cfg.LocalSvrConf, pxy.proxyPlugin, &pxy.cfg.BaseProxyConf, conn,
-		[]byte(pxy.clientCfg.Token), m)
+	HandleTcpWorkConnection(pxy.ctx, &pxy.cfg.LocalSvrConf, pxy.proxyPlugin, &pxy.cfg.BaseProxyConf, pxy.limiter,
+		conn, []byte(pxy.clientCfg.Token), m)
 }
 
 // HTTPS
@@ -185,8 +195,8 @@ func (pxy *HttpsProxy) Close() {
 }
 
 func (pxy *HttpsProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
-	HandleTcpWorkConnection(pxy.ctx, &pxy.cfg.LocalSvrConf, pxy.proxyPlugin, &pxy.cfg.BaseProxyConf, conn,
-		[]byte(pxy.clientCfg.Token), m)
+	HandleTcpWorkConnection(pxy.ctx, &pxy.cfg.LocalSvrConf, pxy.proxyPlugin, &pxy.cfg.BaseProxyConf, pxy.limiter,
+		conn, []byte(pxy.clientCfg.Token), m)
 }
 
 // STCP
@@ -214,8 +224,8 @@ func (pxy *StcpProxy) Close() {
 }
 
 func (pxy *StcpProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
-	HandleTcpWorkConnection(pxy.ctx, &pxy.cfg.LocalSvrConf, pxy.proxyPlugin, &pxy.cfg.BaseProxyConf, conn,
-		[]byte(pxy.clientCfg.Token), m)
+	HandleTcpWorkConnection(pxy.ctx, &pxy.cfg.LocalSvrConf, pxy.proxyPlugin, &pxy.cfg.BaseProxyConf, pxy.limiter,
+		conn, []byte(pxy.clientCfg.Token), m)
 }
 
 // XTCP
@@ -339,7 +349,7 @@ func (pxy *XtcpProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
 
 	lConn.WriteToUDP(sidBuf[:n], uAddr)
 
-	kcpConn, err := frpNet.NewKcpConnFromUdp(lConn, false, natHoleRespMsg.VisitorAddr)
+	kcpConn, err := frpNet.NewKcpConnFromUdp(lConn, false, uAddr.String())
 	if err != nil {
 		xl.Error("create kcp connection from udp connection error: %v", err)
 		return
@@ -360,7 +370,7 @@ func (pxy *XtcpProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
 		return
 	}
 
-	HandleTcpWorkConnection(pxy.ctx, &pxy.cfg.LocalSvrConf, pxy.proxyPlugin, &pxy.cfg.BaseProxyConf,
+	HandleTcpWorkConnection(pxy.ctx, &pxy.cfg.LocalSvrConf, pxy.proxyPlugin, &pxy.cfg.BaseProxyConf, pxy.limiter,
 		muxConn, []byte(pxy.cfg.Sk), m)
 }
 
@@ -429,6 +439,13 @@ func (pxy *UdpProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
 	// close resources releated with old workConn
 	pxy.Close()
 
+	if pxy.limiter != nil {
+		rwc := frpIo.WrapReadWriteCloser(limit.NewReader(conn, pxy.limiter), limit.NewWriter(conn, pxy.limiter), func() error {
+			return conn.Close()
+		})
+		conn = frpNet.WrapReadWriteCloserToConn(rwc, conn)
+	}
+
 	pxy.mu.Lock()
 	pxy.workConn = conn
 	pxy.readCh = make(chan *msg.UdpPacket, 1024)
@@ -491,13 +508,18 @@ func (pxy *UdpProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
 
 // Common handler for tcp work connections.
 func HandleTcpWorkConnection(ctx context.Context, localInfo *config.LocalSvrConf, proxyPlugin plugin.Plugin,
-	baseInfo *config.BaseProxyConf, workConn net.Conn, encKey []byte, m *msg.StartWorkConn) {
+	baseInfo *config.BaseProxyConf, limiter *rate.Limiter, workConn net.Conn, encKey []byte, m *msg.StartWorkConn) {
 	xl := xlog.FromContextSafe(ctx)
 	var (
 		remote io.ReadWriteCloser
 		err    error
 	)
 	remote = workConn
+	if limiter != nil {
+		remote = frpIo.WrapReadWriteCloser(limit.NewReader(workConn, limiter), limit.NewWriter(workConn, limiter), func() error {
+			return workConn.Close()
+		})
+	}
 
 	xl.Trace("handle tcp work connection, use_encryption: %t, use_compression: %t",
 		baseInfo.UseEncryption, baseInfo.UseCompression)

client/service.go

@@ -222,6 +222,7 @@ func (svr *Service) login() (conn net.Conn, session *fmux.Session, err error) {
 		PrivilegeKey: util.GetAuthKey(svr.cfg.Token, now),
 		Timestamp:    now,
 		RunId:        svr.runId,
+		Metas:        svr.cfg.Metas,
 	}
 
 	if err = msg.WriteMsg(conn, loginMsg); err != nil {

conf/frpc_full.ini

@@ -64,6 +64,10 @@ tls_enable = true
 # heartbeat_interval = 30
 # heartbeat_timeout = 90
 
+# additional meta info for client
+meta_var1 = 123
+meta_var2 = 234
+
 # 'ssh' is the unique proxy name
 # if user in [common] section is not empty, it will be changed to {user}.{proxy} such as 'your_name.ssh'
 [ssh]
@@ -71,6 +75,8 @@ tls_enable = true
 type = tcp
 local_ip = 127.0.0.1
 local_port = 22
+# limit bandwith for this proxy, unit is KB and MB
+bandwith_limit = 1MB
 # true or false, if true, messages between frps and frpc will be encrypted, default is false
 use_encryption = false
 # if true, message will be compressed
@@ -90,6 +96,9 @@ health_check_timeout_s = 3
 health_check_max_failed = 3
 # every 10 seconds will do a health check
 health_check_interval_s = 10
+# additional meta info for each proxy
+meta_var1 = 123
+meta_var2 = 234
 
 [ssh_random]
 type = tcp
@@ -205,6 +214,14 @@ plugin_key_path = ./server.key
 plugin_host_header_rewrite = 127.0.0.1
 plugin_header_X-From-Where = frp
 
+[plugin_http2https]
+type = http
+custom_domains = test.yourdomain.com
+plugin = http2https
+plugin_local_addr = 127.0.0.1:443
+plugin_host_header_rewrite = 127.0.0.1
+plugin_header_X-From-Where = frp
+
 [secret_tcp]
 # If the type is secret tcp, remote_port is useless
 # Who want to connect local port should deploy another frpc with stcp proxy and role is visitor

conf/frps_full.ini

@@ -71,3 +71,13 @@ tcp_mux = true
 
 # custom 404 page for HTTP requests
 # custom_404_page = /path/to/404.html
+
+[plugin.user-manager]
+addr = 127.0.0.1:9000
+path = /handler
+ops = Login
+
+[plugin.port-manager]
+addr = 127.0.0.1:9001
+path = /handler
+ops = NewProxy

doc/server_plugin.md

@@ -0,0 +1,171 @@
+### Manage Plugin
+
+frp manage plugin is aim to extend frp's ability without modifing self code.
+
+It runs as a process and listen on a port to provide RPC interface. Before frps doing some operations, frps will send RPC requests to manage plugin and do operations by it's response.
+
+### RPC request
+
+Support HTTP first.
+
+When manage plugin accept the operation request, it can give three different responses.
+
+* Reject operation and return the reason.
+* Allow operation and keep original content.
+* Allow operation and return modified content.
+
+### Interface
+
+HTTP path can be configured for each manage plugin in frps. Assume here is `/handler`.
+
+Request
+
+```
+POST /handler
+{
+    "version": "0.1.0",
+    "op": "Login",
+    "content": {
+        ... // Operation info
+    }
+}
+
+Request Header
+X-Frp-Reqid: for tracing
+```
+
+Response
+
+Error if not return 200 http code.
+
+Reject opeartion
+
+```
+{
+    "reject": true,
+    "reject_reason": "invalid user"
+}
+```
+
+Allow operation and keep original content
+
+```
+{
+    "reject": false,
+    "unchange": true
+}
+```
+
+Allow opeartion and modify content
+
+```
+{
+    "unchange": "false",
+    "content": {
+        ... // Replaced content
+    }
+}
+```
+
+### Operation
+
+Now it supports `Login` and `NewProxy`.
+
+#### Login
+
+Client login operation
+
+```
+{
+    "content": {
+        "version": <string>,
+        "hostname": <string>,
+        "os": <string>,
+        "arch": <string>,
+        "user": <string>,
+        "timestamp": <int64>,
+        "privilege_key": <string>,
+        "run_id": <string>,
+        "pool_count": <int>,
+        "metas": map<string>string
+    }
+}
+```
+
+#### NewProxy
+
+Create new proxy
+
+```
+{
+    "content": {
+        "user": {
+            "user": <string>,
+            "metas": map<string>string
+        },
+        "proxy_name": <string>,
+        "proxy_type": <string>,
+        "use_encryption": <bool>,
+        "use_compression": <bool>,
+        "group": <string>,
+        "group_key": <string>,
+
+        // tcp and udp only
+        "remote_port": <int>,
+
+        // http and https only
+        "custom_domains": []<string>,
+        "subdomain": <string>,
+        "locations": <string>,
+        "http_user": <string>,
+        "http_pwd": <string>,
+        "host_header_rewrite": <string>,
+        "headers": map<string>string,
+
+        "metas": map<string>string
+    }
+}
+```
+
+### manage plugin configure
+
+```ini
+[common]
+bind_port = 7000
+
+[plugin.user-manager]
+addr = 127.0.0.1:9000
+path = /handler
+ops = Login
+
+[plugin.port-manager]
+addr = 127.0.0.1:9001
+path = /handler
+ops = NewProxy
+```
+
+addr: plugin listen on.
+path: http request url path.
+ops: opeartions plugin needs handle.
+
+### meta data
+
+Meta data will be sent to manage plugin in each RCP request.
+
+Meta data start with `meta_`. It can be configured in `common` and each proxy.
+
+```
+# frpc.ini
+[common]
+server_addr = 127.0.0.1
+server_port = 7000
+user = fake
+meta_token = fake
+meta_version = 1.0.0
+
+[ssh]
+type = tcp
+local_port = 22
+remote_port = 6000
+meta_id = 123
+```

doc/server_plugin_zh.md

@@ -0,0 +1,171 @@
+### 服务端管理插件
+
+frp 管理插件的作用是在不侵入自身代码的前提下，扩展 frp 服务端的能力。
+
+frp 管理插件会以单独进程的形式运行，并且监听在一个端口上，对外提供 RPC 接口，响应 frps 的请求。
+
+frps 在执行某些操作前，会根据配置向管理插件发送 RPC 请求，根据管理插件的响应来执行相应的操作。
+
+### RPC 请求
+
+管理插件接收到操作请求后，可以给出三种回应。
+
+* 拒绝操作，需要返回拒绝操作的原因。
+* 允许操作，不需要修改操作内容。
+* 允许操作，对操作请求进行修改后，返回修改后的内容。
+
+### 接口
+
+接口路径可以在 frps 配置中为每个插件单独配置，这里以 `/handler` 为例。
+
+Request
+
+```
+POST /handler
+{
+    "version": "0.1.0",
+    "op": "Login",
+    "content": {
+        ... // 具体的操作信息
+    }
+}
+
+请求 Header
+X-Frp-Reqid: 用于追踪请求
+```
+
+Response
+
+非 200 的返回都认为是请求异常。
+
+拒绝执行操作
+
+```
+{
+   "reject": true,
+   "reject_reason": "invalid user"
+}
+```
+
+允许且内容不需要变动
+
+```
+{
+    "reject": false,
+    "unchange": true
+}
+```
+
+允许且需要替换操作内容
+
+```
+{
+    "unchange": "false",
+    "content": {
+        ... // 替换后的操作信息，格式必须和请求时的一致
+    }
+}
+```
+
+### 操作类型
+
+目前插件支持管理的操作类型有 `Login` 和 `NewProxy`。
+
+#### Login
+
+用户登录操作信息
+
+```
+{
+    "content": {
+        "version": <string>,
+        "hostname": <string>,
+        "os": <string>,
+        "arch": <string>,
+        "user": <string>,
+        "timestamp": <int64>,
+        "privilege_key": <string>,
+        "run_id": <string>,
+        "pool_count": <int>,
+        "metas": map<string>string
+    }
+}
+```
+
+#### NewProxy
+
+创建代理的相关信息
+
+```
+{
+    "content": {
+        "user": {
+            "user": <string>,
+            "metas": map<string>string
+        },
+        "proxy_name": <string>,
+        "proxy_type": <string>,
+        "use_encryption": <bool>,
+        "use_compression": <bool>,
+        "group": <string>,
+        "group_key": <string>,
+
+        // tcp and udp only
+        "remote_port": <int>,
+
+        // http and https only
+        "custom_domains": []<string>,
+        "subdomain": <string>,
+        "locations": <string>,
+        "http_user": <string>,
+        "http_pwd": <string>,
+        "host_header_rewrite": <string>,
+        "headers": map<string>string,
+
+        "metas": map<string>string
+    }
+}
+```
+
+### frps 中插件配置
+
+```ini
+[common]
+bind_port = 7000
+
+[plugin.user-manager]
+addr = 127.0.0.1:9000
+path = /handler
+ops = Login
+
+[plugin.port-manager]
+addr = 127.0.0.1:9001
+path = /handler
+ops = NewProxy
+```
+
+addr: 插件监听的网络地址。
+path: 插件监听的 HTTP 请求路径。
+ops: 插件需要处理的操作列表，多个 op 以英文逗号分隔。
+
+### 元数据
+
+为了减少 frps 的代码修改，同时提高管理插件的扩展能力，在 frpc 的配置文件中引入自定义元数据的概念。元数据会在调用 RPC 请求时发送给插件。
+
+元数据以 `meta_` 开头，可以配置多个，元数据分为两种，一种配置在 `common` 下，一种配置在各个 proxy 中。
+
+```
+# frpc.ini
+[common]
+server_addr = 127.0.0.1
+server_port = 7000
+user = fake
+meta_token = fake
+meta_version = 1.0.0
+
+[ssh]
+type = tcp
+local_port = 22
+remote_port = 6000
+meta_id = 123
+```

go.mod

@@ -29,4 +29,5 @@ require (
 	github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae // indirect
 	golang.org/x/net v0.0.0-20190724013045-ca1201d0de80
 	golang.org/x/text v0.3.2 // indirect
+	golang.org/x/time v0.0.0-20191024005414-555d28b269f0
 )

go.sum

@@ -46,4 +46,6 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/time v0.0.0-20191024005414-555d28b269f0 h1:/5xXl8Y5W96D+TtHSlonuFqGHIWVuyCkGJLwGh9JJFs=
+golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

models/config/client_common.go

@@ -115,6 +115,8 @@ type ClientCommonConf struct {
 	// before the connection is terminated, in seconds. It is not recommended
 	// to change this value. By default, this value is 90.
 	HeartBeatTimeout int64 `json:"heartbeat_timeout"`
+	// Client meta info
+	Metas map[string]string `json:"metas"`
 }
 
 // GetDefaultClientConf returns a client configuration with default values.
@@ -144,6 +146,7 @@ func GetDefaultClientConf() ClientCommonConf {
 		TLSEnable:         false,
 		HeartBeatInterval: 30,
 		HeartBeatTimeout:  90,
+		Metas:             make(map[string]string),
 	}
 }
 
@@ -294,6 +297,11 @@ func UnmarshalClientConfFromIni(content string) (cfg ClientCommonConf, err error
 			cfg.HeartBeatInterval = v
 		}
 	}
+	for k, v := range conf.Section("common") {
+		if strings.HasPrefix(k, "meta_") {
+			cfg.Metas[strings.TrimPrefix(k, "meta_")] = v
+		}
+	}
 	return
 }
 

models/config/proxy.go

@@ -125,6 +125,14 @@ type BaseProxyConf struct {
 	// values include "v1", "v2", and "". If the value is "", a protocol
 	// version will be automatically selected. By default, this value is "".
 	ProxyProtocolVersion string `json:"proxy_protocol_version"`
+
+	// BandwidthLimit limit the proxy bandwidth
+	// 0 means no limit
+	BandwidthLimit BandwidthQuantity `json:"bandwidth_limit"`
+
+	// meta info for each proxy
+	Metas map[string]string `json:"metas"`
+
 	LocalSvrConf
 	HealthCheckConf
 }
@@ -140,7 +148,9 @@ func (cfg *BaseProxyConf) compare(cmp *BaseProxyConf) bool {
 		cfg.UseCompression != cmp.UseCompression ||
 		cfg.Group != cmp.Group ||
 		cfg.GroupKey != cmp.GroupKey ||
-		cfg.ProxyProtocolVersion != cmp.ProxyProtocolVersion {
+		cfg.ProxyProtocolVersion != cmp.ProxyProtocolVersion ||
+		cfg.BandwidthLimit.Equal(&cmp.BandwidthLimit) ||
+		!reflect.DeepEqual(cfg.Metas, cmp.Metas) {
 		return false
 	}
 	if !cfg.LocalSvrConf.compare(&cmp.LocalSvrConf) {
@@ -159,12 +169,14 @@ func (cfg *BaseProxyConf) UnmarshalFromMsg(pMsg *msg.NewProxy) {
 	cfg.UseCompression = pMsg.UseCompression
 	cfg.Group = pMsg.Group
 	cfg.GroupKey = pMsg.GroupKey
+	cfg.Metas = pMsg.Metas
 }
 
 func (cfg *BaseProxyConf) UnmarshalFromIni(prefix string, name string, section ini.Section) error {
 	var (
 		tmpStr string
 		ok     bool
+		err    error
 	)
 	cfg.ProxyName = prefix + name
 	cfg.ProxyType = section["type"]
@@ -183,11 +195,15 @@ func (cfg *BaseProxyConf) UnmarshalFromIni(prefix string, name string, section i
 	cfg.GroupKey = section["group_key"]
 	cfg.ProxyProtocolVersion = section["proxy_protocol_version"]
 
-	if err := cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil {
+	if cfg.BandwidthLimit, err = NewBandwidthQuantity(section["bandwidth_limit"]); err != nil {
 		return err
 	}
 
-	if err := cfg.HealthCheckConf.UnmarshalFromIni(prefix, name, section); err != nil {
+	if err = cfg.LocalSvrConf.UnmarshalFromIni(prefix, name, section); err != nil {
+		return err
+	}
+
+	if err = cfg.HealthCheckConf.UnmarshalFromIni(prefix, name, section); err != nil {
 		return err
 	}
 
@@ -201,6 +217,12 @@ func (cfg *BaseProxyConf) UnmarshalFromIni(prefix string, name string, section i
 		}
 		cfg.HealthCheckUrl = s + cfg.HealthCheckUrl
 	}
+
+	for k, v := range section {
+		if strings.HasPrefix(k, "meta_") {
+			cfg.Metas[strings.TrimPrefix(k, "meta_")] = v
+		}
+	}
 	return nil
 }
 
@@ -211,6 +233,7 @@ func (cfg *BaseProxyConf) MarshalToMsg(pMsg *msg.NewProxy) {
 	pMsg.UseCompression = cfg.UseCompression
 	pMsg.Group = cfg.Group
 	pMsg.GroupKey = cfg.GroupKey
+	pMsg.Metas = cfg.Metas
 }
 
 func (cfg *BaseProxyConf) checkForCli() (err error) {

models/config/server_common.go

@@ -21,6 +21,7 @@ import (
 
 	ini "github.com/vaughan0/go-ini"
 
+	plugin "github.com/fatedier/frp/models/plugin/server"
 	"github.com/fatedier/frp/utils/util"
 )
 
@@ -134,6 +135,8 @@ type ServerCommonConf struct {
 	// UserConnTimeout specifies the maximum time to wait for a work
 	// connection. By default, this value is 10.
 	UserConnTimeout int64 `json:"user_conn_timeout"`
+	// HTTPPlugins specify the server plugins support HTTP protocol.
+	HTTPPlugins map[string]plugin.HTTPPluginOptions `json:"http_plugins"`
 }
 
 // GetDefaultServerConf returns a server configuration with reasonable
@@ -167,6 +170,7 @@ func GetDefaultServerConf() ServerCommonConf {
 		HeartBeatTimeout:  90,
 		UserConnTimeout:   10,
 		Custom404Page:     "",
+		HTTPPlugins:       make(map[string]plugin.HTTPPluginOptions),
 	}
 }
 
@@ -181,6 +185,8 @@ func UnmarshalServerConfFromIni(content string) (cfg ServerCommonConf, err error
 		return ServerCommonConf{}, err
 	}
 
+	UnmarshalPluginsFromIni(conf, &cfg)
+
 	var (
 		tmpStr string
 		ok     bool
@@ -375,6 +381,24 @@ func UnmarshalServerConfFromIni(content string) (cfg ServerCommonConf, err error
 	return
 }
 
+func UnmarshalPluginsFromIni(sections ini.File, cfg *ServerCommonConf) {
+	for name, section := range sections {
+		if strings.HasPrefix(name, "plugin.") {
+			name = strings.TrimSpace(strings.TrimPrefix(name, "plugin."))
+			options := plugin.HTTPPluginOptions{
+				Name: name,
+				Addr: section["addr"],
+				Path: section["path"],
+				Ops:  strings.Split(section["ops"], ","),
+			}
+			for i, _ := range options.Ops {
+				options.Ops[i] = strings.TrimSpace(options.Ops[i])
+			}
+			cfg.HTTPPlugins[name] = options
+		}
+	}
+}
+
 func (cfg *ServerCommonConf) Check() (err error) {
 	return
 }

models/config/types.go

@@ -0,0 +1,112 @@
+// Copyright 2019 fatedier, fatedier@gmail.com
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"encoding/json"
+	"errors"
+	"strconv"
+	"strings"
+)
+
+const (
+	MB = 1024 * 1024
+	KB = 1024
+)
+
+type BandwidthQuantity struct {
+	s string // MB or KB
+
+	i int64 // bytes
+}
+
+func NewBandwidthQuantity(s string) (BandwidthQuantity, error) {
+	q := BandwidthQuantity{}
+	err := q.UnmarshalString(s)
+	if err != nil {
+		return q, err
+	}
+	return q, nil
+}
+
+func (q *BandwidthQuantity) Equal(u *BandwidthQuantity) bool {
+	if q == nil && u == nil {
+		return true
+	}
+	if q != nil && u != nil {
+		return q.i == u.i
+	}
+	return false
+}
+
+func (q *BandwidthQuantity) String() string {
+	return q.s
+}
+
+func (q *BandwidthQuantity) UnmarshalString(s string) error {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return nil
+	}
+
+	var (
+		base int64
+		f    float64
+		err  error
+	)
+	if strings.HasSuffix(s, "MB") {
+		base = MB
+		fstr := strings.TrimSuffix(s, "MB")
+		f, err = strconv.ParseFloat(fstr, 64)
+		if err != nil {
+			return err
+		}
+	} else if strings.HasSuffix(s, "KB") {
+		base = KB
+		fstr := strings.TrimSuffix(s, "KB")
+		f, err = strconv.ParseFloat(fstr, 64)
+		if err != nil {
+			return err
+		}
+	} else {
+		return errors.New("unit not support")
+	}
+
+	q.s = s
+	q.i = int64(f * float64(base))
+	return nil
+}
+
+func (q *BandwidthQuantity) UnmarshalJSON(b []byte) error {
+	if len(b) == 4 && string(b) == "null" {
+		return nil
+	}
+
+	var str string
+	err := json.Unmarshal(b, &str)
+	if err != nil {
+		return err
+	}
+
+	return q.UnmarshalString(str)
+}
+
+func (q *BandwidthQuantity) MarshalJSON() ([]byte, error) {
+	return []byte("\"" + q.s + "\""), nil
+}
+
+func (q *BandwidthQuantity) Bytes() int64 {
+	return q.i
+}

models/config/types_test.go

@@ -0,0 +1,40 @@
+// Copyright 2019 fatedier, fatedier@gmail.com
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type Wrap struct {
+	B   BandwidthQuantity `json:"b"`
+	Int int               `json:"int"`
+}
+
+func TestBandwidthQuantity(t *testing.T) {
+	assert := assert.New(t)
+
+	var w Wrap
+	err := json.Unmarshal([]byte(`{"b":"1KB","int":5}`), &w)
+	assert.NoError(err)
+	assert.EqualValues(1*KB, w.B.Bytes())
+
+	buf, err := json.Marshal(&w)
+	assert.NoError(err)
+	assert.Equal(`{"b":"1KB","int":5}`, string(buf))
+}

models/msg/msg.go

@@ -62,14 +62,15 @@ var (
 
 // When frpc start, client send this message to login to server.
 type Login struct {
-	Version      string `json:"version"`
-	Hostname     string `json:"hostname"`
-	Os           string `json:"os"`
-	Arch         string `json:"arch"`
-	User         string `json:"user"`
-	PrivilegeKey string `json:"privilege_key"`
-	Timestamp    int64  `json:"timestamp"`
-	RunId        string `json:"run_id"`
+	Version      string            `json:"version"`
+	Hostname     string            `json:"hostname"`
+	Os           string            `json:"os"`
+	Arch         string            `json:"arch"`
+	User         string            `json:"user"`
+	PrivilegeKey string            `json:"privilege_key"`
+	Timestamp    int64             `json:"timestamp"`
+	RunId        string            `json:"run_id"`
+	Metas        map[string]string `json:"metas"`
 
 	// Some global configures.
 	PoolCount int `json:"pool_count"`
@@ -84,12 +85,13 @@ type LoginResp struct {
 
 // When frpc login success, send this message to frps for running a new proxy.
 type NewProxy struct {
-	ProxyName      string `json:"proxy_name"`
-	ProxyType      string `json:"proxy_type"`
-	UseEncryption  bool   `json:"use_encryption"`
-	UseCompression bool   `json:"use_compression"`
-	Group          string `json:"group"`
-	GroupKey       string `json:"group_key"`
+	ProxyName      string            `json:"proxy_name"`
+	ProxyType      string            `json:"proxy_type"`
+	UseEncryption  bool              `json:"use_encryption"`
+	UseCompression bool              `json:"use_compression"`
+	Group          string            `json:"group"`
+	GroupKey       string            `json:"group_key"`
+	Metas          map[string]string `json:"metas"`
 
 	// tcp and udp only
 	RemotePort int `json:"remote_port"`

models/plugin/client/http2https.go

@@ -0,0 +1,111 @@
+// Copyright 2019 fatedier, fatedier@gmail.com
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package plugin
+
+import (
+	"crypto/tls"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"strings"
+
+	frpNet "github.com/fatedier/frp/utils/net"
+)
+
+const PluginHTTP2HTTPS = "http2https"
+
+func init() {
+	Register(PluginHTTP2HTTPS, NewHTTP2HTTPSPlugin)
+}
+
+type HTTP2HTTPSPlugin struct {
+	hostHeaderRewrite string
+	localAddr         string
+	headers           map[string]string
+
+	l *Listener
+	s *http.Server
+}
+
+func NewHTTP2HTTPSPlugin(params map[string]string) (Plugin, error) {
+	localAddr := params["plugin_local_addr"]
+	hostHeaderRewrite := params["plugin_host_header_rewrite"]
+	headers := make(map[string]string)
+	for k, v := range params {
+		if !strings.HasPrefix(k, "plugin_header_") {
+			continue
+		}
+		if k = strings.TrimPrefix(k, "plugin_header_"); k != "" {
+			headers[k] = v
+		}
+	}
+
+	if localAddr == "" {
+		return nil, fmt.Errorf("plugin_local_addr is required")
+	}
+
+	listener := NewProxyListener()
+
+	p := &HTTPS2HTTPPlugin{
+		localAddr:         localAddr,
+		hostHeaderRewrite: hostHeaderRewrite,
+		headers:           headers,
+		l:                 listener,
+	}
+
+	tr := &http.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+	}
+
+	rp := &httputil.ReverseProxy{
+		Director: func(req *http.Request) {
+			req.URL.Scheme = "https"
+			req.URL.Host = p.localAddr
+			if p.hostHeaderRewrite != "" {
+				req.Host = p.hostHeaderRewrite
+			}
+			for k, v := range p.headers {
+				req.Header.Set(k, v)
+			}
+		},
+		Transport: tr,
+	}
+
+	p.s = &http.Server{
+		Handler: rp,
+	}
+
+	go p.s.Serve(listener)
+
+	return p, nil
+}
+
+func (p *HTTP2HTTPSPlugin) Handle(conn io.ReadWriteCloser, realConn net.Conn, extraBufToLocal []byte) {
+	wrapConn := frpNet.WrapReadWriteCloserToConn(conn, realConn)
+	p.l.PutConn(wrapConn)
+}
+
+func (p *HTTP2HTTPSPlugin) Name() string {
+	return PluginHTTP2HTTPS
+}
+
+func (p *HTTP2HTTPSPlugin) Close() error {
+	if err := p.s.Close(); err != nil {
+		return err
+	}
+	return nil
+}

models/plugin/client/https2http.go

@@ -126,5 +126,8 @@ func (p *HTTPS2HTTPPlugin) Name() string {
 }
 
 func (p *HTTPS2HTTPPlugin) Close() error {
+	if err := p.s.Close(); err != nil {
+		return err
+	}
 	return nil
 }

models/plugin/server/http.go

@@ -0,0 +1,104 @@
+// Copyright 2019 fatedier, fatedier@gmail.com
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package plugin
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"reflect"
+)
+
+type HTTPPluginOptions struct {
+	Name string
+	Addr string
+	Path string
+	Ops  []string
+}
+
+type httpPlugin struct {
+	options HTTPPluginOptions
+
+	url    string
+	client *http.Client
+}
+
+func NewHTTPPluginOptions(options HTTPPluginOptions) Plugin {
+	return &httpPlugin{
+		options: options,
+		url:     fmt.Sprintf("http://%s%s", options.Addr, options.Path),
+		client:  &http.Client{},
+	}
+}
+
+func (p *httpPlugin) Name() string {
+	return p.options.Name
+}
+
+func (p *httpPlugin) IsSupport(op string) bool {
+	for _, v := range p.options.Ops {
+		if v == op {
+			return true
+		}
+	}
+	return false
+}
+
+func (p *httpPlugin) Handle(ctx context.Context, op string, content interface{}) (*Response, interface{}, error) {
+	r := &Request{
+		Version: APIVersion,
+		Op:      op,
+		Content: content,
+	}
+	var res Response
+	res.Content = reflect.New(reflect.TypeOf(content)).Interface()
+	if err := p.do(ctx, r, &res); err != nil {
+		return nil, nil, err
+	}
+	return &res, res.Content, nil
+}
+
+func (p *httpPlugin) do(ctx context.Context, r *Request, res *Response) error {
+	buf, err := json.Marshal(r)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequest("POST", p.url, bytes.NewReader(buf))
+	if err != nil {
+		return err
+	}
+	req = req.WithContext(ctx)
+	req.Header.Set("X-Frp-Reqid", GetReqidFromContext(ctx))
+	resp, err := p.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("do http request error code: %d", resp.StatusCode)
+	}
+	buf, err = ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+	if err = json.Unmarshal(buf, res); err != nil {
+		return err
+	}
+	return nil
+}

models/plugin/server/manager.go

@@ -0,0 +1,105 @@
+// Copyright 2019 fatedier, fatedier@gmail.com
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package plugin
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/fatedier/frp/utils/util"
+	"github.com/fatedier/frp/utils/xlog"
+)
+
+type Manager struct {
+	loginPlugins    []Plugin
+	newProxyPlugins []Plugin
+}
+
+func NewManager() *Manager {
+	return &Manager{
+		loginPlugins:    make([]Plugin, 0),
+		newProxyPlugins: make([]Plugin, 0),
+	}
+}
+
+func (m *Manager) Register(p Plugin) {
+	if p.IsSupport(OpLogin) {
+		m.loginPlugins = append(m.loginPlugins, p)
+	}
+	if p.IsSupport(OpNewProxy) {
+		m.newProxyPlugins = append(m.newProxyPlugins, p)
+	}
+}
+
+func (m *Manager) Login(content *LoginContent) (*LoginContent, error) {
+	var (
+		res = &Response{
+			Reject:   false,
+			Unchange: true,
+		}
+		retContent interface{}
+		err        error
+	)
+	reqid, _ := util.RandId()
+	xl := xlog.New().AppendPrefix("reqid: " + reqid)
+	ctx := xlog.NewContext(context.Background(), xl)
+	ctx = NewReqidContext(ctx, reqid)
+
+	for _, p := range m.loginPlugins {
+		res, retContent, err = p.Handle(ctx, OpLogin, *content)
+		if err != nil {
+			xl.Warn("send Login request to plugin [%s] error: %v", p.Name(), err)
+			return nil, errors.New("send Login request to plugin error")
+		}
+		if res.Reject {
+			return nil, fmt.Errorf("%s", res.RejectReason)
+		}
+		if !res.Unchange {
+			content = retContent.(*LoginContent)
+		}
+	}
+	return content, nil
+}
+
+func (m *Manager) NewProxy(content *NewProxyContent) (*NewProxyContent, error) {
+	var (
+		res = &Response{
+			Reject:   false,
+			Unchange: true,
+		}
+		retContent interface{}
+		err        error
+	)
+	reqid, _ := util.RandId()
+	xl := xlog.New().AppendPrefix("reqid: " + reqid)
+	ctx := xlog.NewContext(context.Background(), xl)
+	ctx = NewReqidContext(ctx, reqid)
+
+	for _, p := range m.newProxyPlugins {
+		res, retContent, err = p.Handle(ctx, OpNewProxy, *content)
+		if err != nil {
+			xl.Warn("send NewProxy request to plugin [%s] error: %v", p.Name(), err)
+			return nil, errors.New("send NewProxy request to plugin error")
+		}
+		if res.Reject {
+			return nil, fmt.Errorf("%s", res.RejectReason)
+		}
+		if !res.Unchange {
+			content = retContent.(*NewProxyContent)
+		}
+	}
+	return content, nil
+}

models/plugin/server/plugin.go

@@ -0,0 +1,32 @@
+// Copyright 2019 fatedier, fatedier@gmail.com
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package plugin
+
+import (
+	"context"
+)
+
+const (
+	APIVersion = "0.1.0"
+
+	OpLogin    = "Login"
+	OpNewProxy = "NewProxy"
+)
+
+type Plugin interface {
+	Name() string
+	IsSupport(op string) bool
+	Handle(ctx context.Context, op string, content interface{}) (res *Response, retContent interface{}, err error)
+}

models/plugin/server/tracer.go

@@ -0,0 +1,34 @@
+// Copyright 2019 fatedier, fatedier@gmail.com
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package plugin
+
+import (
+	"context"
+)
+
+type key int
+
+const (
+	reqidKey key = 0
+)
+
+func NewReqidContext(ctx context.Context, reqid string) context.Context {
+	return context.WithValue(ctx, reqidKey, reqid)
+}
+
+func GetReqidFromContext(ctx context.Context) string {
+	ret, _ := ctx.Value(reqidKey).(string)
+	return ret
+}

models/plugin/server/types.go

@@ -0,0 +1,46 @@
+// Copyright 2019 fatedier, fatedier@gmail.com
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package plugin
+
+import (
+	"github.com/fatedier/frp/models/msg"
+)
+
+type Request struct {
+	Version string      `json:"version"`
+	Op      string      `json:"op"`
+	Content interface{} `json:"content"`
+}
+
+type Response struct {
+	Reject       bool        `json:"reject"`
+	RejectReason string      `json:"reject_reason"`
+	Unchange     bool        `json:"unchange"`
+	Content      interface{} `json:"content"`
+}
+
+type LoginContent struct {
+	msg.Login
+}
+
+type UserInfo struct {
+	User  string            `json:"user"`
+	Metas map[string]string `json:"metas"`
+}
+
+type NewProxyContent struct {
+	User UserInfo `json:"user"`
+	msg.NewProxy
+}

server/control.go

@@ -27,6 +27,7 @@ import (
 	"github.com/fatedier/frp/models/consts"
 	frpErr "github.com/fatedier/frp/models/errors"
 	"github.com/fatedier/frp/models/msg"
+	plugin "github.com/fatedier/frp/models/plugin/server"
 	"github.com/fatedier/frp/server/controller"
 	"github.com/fatedier/frp/server/proxy"
 	"github.com/fatedier/frp/server/stats"
@@ -86,6 +87,9 @@ type Control struct {
 	// proxy manager
 	pxyManager *proxy.ProxyManager
 
+	// plugin manager
+	pluginManager *plugin.Manager
+
 	// stats collector to store stats info of clients and proxies
 	statsCollector stats.Collector
 
@@ -138,9 +142,16 @@ type Control struct {
 	ctx context.Context
 }
 
-func NewControl(ctx context.Context, rc *controller.ResourceController, pxyManager *proxy.ProxyManager,
-	statsCollector stats.Collector, ctlConn net.Conn, loginMsg *msg.Login,
-	serverCfg config.ServerCommonConf) *Control {
+func NewControl(
+	ctx context.Context,
+	rc *controller.ResourceController,
+	pxyManager *proxy.ProxyManager,
+	pluginManager *plugin.Manager,
+	statsCollector stats.Collector,
+	ctlConn net.Conn,
+	loginMsg *msg.Login,
+	serverCfg config.ServerCommonConf,
+) *Control {
 
 	poolCount := loginMsg.PoolCount
 	if poolCount > int(serverCfg.MaxPoolCount) {
@@ -149,6 +160,7 @@ func NewControl(ctx context.Context, rc *controller.ResourceController, pxyManag
 	return &Control{
 		rc:              rc,
 		pxyManager:      pxyManager,
+		pluginManager:   pluginManager,
 		statsCollector:  statsCollector,
 		conn:            ctlConn,
 		loginMsg:        loginMsg,
@@ -407,8 +419,21 @@ func (ctl *Control) manager() {
 
 			switch m := rawMsg.(type) {
 			case *msg.NewProxy:
+				content := &plugin.NewProxyContent{
+					User: plugin.UserInfo{
+						User:  ctl.loginMsg.User,
+						Metas: ctl.loginMsg.Metas,
+					},
+					NewProxy: *m,
+				}
+				var remoteAddr string
+				retContent, err := ctl.pluginManager.NewProxy(content)
+				if err == nil {
+					m = &retContent.NewProxy
+					remoteAddr, err = ctl.RegisterProxy(m)
+				}
+
 				// register proxy in this control
-				remoteAddr, err := ctl.RegisterProxy(m)
 				resp := &msg.NewProxyResp{
 					ProxyName: m.ProxyName,
 				}

server/service.go

@@ -33,6 +33,7 @@ import (
 	"github.com/fatedier/frp/models/config"
 	"github.com/fatedier/frp/models/msg"
 	"github.com/fatedier/frp/models/nathole"
+	plugin "github.com/fatedier/frp/models/plugin/server"
 	"github.com/fatedier/frp/server/controller"
 	"github.com/fatedier/frp/server/group"
 	"github.com/fatedier/frp/server/ports"
@@ -76,6 +77,9 @@ type Service struct {
 	// Manage all proxies
 	pxyManager *proxy.ProxyManager
 
+	// Manage all plugins
+	pluginManager *plugin.Manager
+
 	// HTTP vhost router
 	httpVhostRouter *vhost.VhostRouters
 
@@ -92,8 +96,9 @@ type Service struct {
 
 func NewService(cfg config.ServerCommonConf) (svr *Service, err error) {
 	svr = &Service{
-		ctlManager: NewControlManager(),
-		pxyManager: proxy.NewProxyManager(),
+		ctlManager:    NewControlManager(),
+		pxyManager:    proxy.NewProxyManager(),
+		pluginManager: plugin.NewManager(),
 		rc: &controller.ResourceController{
 			VisitorManager: controller.NewVisitorManager(),
 			TcpPortManager: ports.NewPortManager("tcp", cfg.ProxyBindAddr, cfg.AllowPorts),
@@ -104,6 +109,12 @@ func NewService(cfg config.ServerCommonConf) (svr *Service, err error) {
 		cfg:             cfg,
 	}
 
+	// Init all plugins
+	for name, options := range cfg.HTTPPlugins {
+		svr.pluginManager.Register(plugin.NewHTTPPluginOptions(options))
+		log.Info("plugin [%s] has been registered", name)
+	}
+
 	// Init group controller
 	svr.rc.TcpGroupCtl = group.NewTcpGroupCtl(svr.rc.TcpPortManager)
 
@@ -295,7 +306,16 @@ func (svr *Service) HandleListener(l net.Listener) {
 
 				switch m := rawMsg.(type) {
 				case *msg.Login:
-					err = svr.RegisterControl(conn, m)
+					// server plugin hook
+					content := &plugin.LoginContent{
+						Login: *m,
+					}
+					retContent, err := svr.pluginManager.Login(content)
+					if err == nil {
+						m = &retContent.Login
+						err = svr.RegisterControl(conn, m)
+					}
+
 					// If login failed, send error message there.
 					// Otherwise send success message in control's work goroutine.
 					if err != nil {
@@ -384,7 +404,7 @@ func (svr *Service) RegisterControl(ctlConn net.Conn, loginMsg *msg.Login) (err
 		return
 	}
 
-	ctl := NewControl(ctx, svr.rc, svr.pxyManager, svr.statsCollector, ctlConn, loginMsg, svr.cfg)
+	ctl := NewControl(ctx, svr.rc, svr.pxyManager, svr.pluginManager, svr.statsCollector, ctlConn, loginMsg, svr.cfg)
 
 	if oldCtl := svr.ctlManager.Add(loginMsg.RunId, ctl); oldCtl != nil {
 		oldCtl.allShutdown.WaitDone()

utils/limit/reader.go

@@ -0,0 +1,51 @@
+// Copyright 2019 fatedier, fatedier@gmail.com
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package limit
+
+import (
+	"context"
+	"io"
+
+	"golang.org/x/time/rate"
+)
+
+type Reader struct {
+	r       io.Reader
+	limiter *rate.Limiter
+}
+
+func NewReader(r io.Reader, limiter *rate.Limiter) *Reader {
+	return &Reader{
+		r:       r,
+		limiter: limiter,
+	}
+}
+
+func (r *Reader) Read(p []byte) (n int, err error) {
+	b := r.limiter.Burst()
+	if b < len(p) {
+		p = p[:b]
+	}
+	n, err = r.r.Read(p)
+	if err != nil {
+		return
+	}
+
+	err = r.limiter.WaitN(context.Background(), n)
+	if err != nil {
+		return
+	}
+	return
+}

utils/limit/writer.go

@@ -0,0 +1,60 @@
+// Copyright 2019 fatedier, fatedier@gmail.com
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package limit
+
+import (
+	"context"
+	"io"
+
+	"golang.org/x/time/rate"
+)
+
+type Writer struct {
+	w       io.Writer
+	limiter *rate.Limiter
+}
+
+func NewWriter(w io.Writer, limiter *rate.Limiter) *Writer {
+	return &Writer{
+		w:       w,
+		limiter: limiter,
+	}
+}
+
+func (w *Writer) Write(p []byte) (n int, err error) {
+	var nn int
+	b := w.limiter.Burst()
+	for {
+		end := len(p)
+		if end == 0 {
+			break
+		}
+		if b < len(p) {
+			end = b
+		}
+		err = w.limiter.WaitN(context.Background(), end)
+		if err != nil {
+			return
+		}
+
+		nn, err = w.w.Write(p[:end])
+		n += nn
+		if err != nil {
+			return
+		}
+		p = p[end:]
+	}
+	return
+}

utils/version/version.go

@@ -19,7 +19,7 @@ import (
 	"strings"
 )
 
-var version string = "0.29.1"
+var version string = "0.31.0"
 
 func Full() string {
 	return version

vendor/golang.org/x/time/AUTHORS

@@ -0,0 +1,3 @@
+# This source code refers to The Go Authors for copyright purposes.
+# The master list of authors is in the main Go distribution,
+# visible at http://tip.golang.org/AUTHORS.

vendor/golang.org/x/time/CONTRIBUTORS

@@ -0,0 +1,3 @@
+# This source code was written by the Go contributors.
+# The master list of contributors is in the main Go distribution,
+# visible at http://tip.golang.org/CONTRIBUTORS.

vendor/golang.org/x/time/LICENSE

@@ -0,0 +1,27 @@
+Copyright (c) 2009 The Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+   * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+   * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

vendor/golang.org/x/time/PATENTS

@@ -0,0 +1,22 @@
+Additional IP Rights Grant (Patents)
+
+"This implementation" means the copyrightable works distributed by
+Google as part of the Go project.
+
+Google hereby grants to You a perpetual, worldwide, non-exclusive,
+no-charge, royalty-free, irrevocable (except as stated in this section)
+patent license to make, have made, use, offer to sell, sell, import,
+transfer and otherwise run, modify and propagate the contents of this
+implementation of Go, where such license applies only to those patent
+claims, both currently owned or controlled by Google and acquired in
+the future, licensable by Google that are necessarily infringed by this
+implementation of Go.  This grant does not include claims that would be
+infringed only as a consequence of further modification of this
+implementation.  If you or your agent or exclusive licensee institute or
+order or agree to the institution of patent litigation against any
+entity (including a cross-claim or counterclaim in a lawsuit) alleging
+that this implementation of Go or any code incorporated within this
+implementation of Go constitutes direct or contributory patent
+infringement, or inducement of patent infringement, then any patent
+rights granted to you under this License for this implementation of Go
+shall terminate as of the date such litigation is filed.

vendor/golang.org/x/time/rate/rate.go

@@ -0,0 +1,400 @@
+// Copyright 2015 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package rate provides a rate limiter.
+package rate
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"sync"
+	"time"
+)
+
+// Limit defines the maximum frequency of some events.
+// Limit is represented as number of events per second.
+// A zero Limit allows no events.
+type Limit float64
+
+// Inf is the infinite rate limit; it allows all events (even if burst is zero).
+const Inf = Limit(math.MaxFloat64)
+
+// Every converts a minimum time interval between events to a Limit.
+func Every(interval time.Duration) Limit {
+	if interval <= 0 {
+		return Inf
+	}
+	return 1 / Limit(interval.Seconds())
+}
+
+// A Limiter controls how frequently events are allowed to happen.
+// It implements a "token bucket" of size b, initially full and refilled
+// at rate r tokens per second.
+// Informally, in any large enough time interval, the Limiter limits the
+// rate to r tokens per second, with a maximum burst size of b events.
+// As a special case, if r == Inf (the infinite rate), b is ignored.
+// See https://en.wikipedia.org/wiki/Token_bucket for more about token buckets.
+//
+// The zero value is a valid Limiter, but it will reject all events.
+// Use NewLimiter to create non-zero Limiters.
+//
+// Limiter has three main methods, Allow, Reserve, and Wait.
+// Most callers should use Wait.
+//
+// Each of the three methods consumes a single token.
+// They differ in their behavior when no token is available.
+// If no token is available, Allow returns false.
+// If no token is available, Reserve returns a reservation for a future token
+// and the amount of time the caller must wait before using it.
+// If no token is available, Wait blocks until one can be obtained
+// or its associated context.Context is canceled.
+//
+// The methods AllowN, ReserveN, and WaitN consume n tokens.
+type Limiter struct {
+	limit Limit
+	burst int
+
+	mu     sync.Mutex
+	tokens float64
+	// last is the last time the limiter's tokens field was updated
+	last time.Time
+	// lastEvent is the latest time of a rate-limited event (past or future)
+	lastEvent time.Time
+}
+
+// Limit returns the maximum overall event rate.
+func (lim *Limiter) Limit() Limit {
+	lim.mu.Lock()
+	defer lim.mu.Unlock()
+	return lim.limit
+}
+
+// Burst returns the maximum burst size. Burst is the maximum number of tokens
+// that can be consumed in a single call to Allow, Reserve, or Wait, so higher
+// Burst values allow more events to happen at once.
+// A zero Burst allows no events, unless limit == Inf.
+func (lim *Limiter) Burst() int {
+	return lim.burst
+}
+
+// NewLimiter returns a new Limiter that allows events up to rate r and permits
+// bursts of at most b tokens.
+func NewLimiter(r Limit, b int) *Limiter {
+	return &Limiter{
+		limit: r,
+		burst: b,
+	}
+}
+
+// Allow is shorthand for AllowN(time.Now(), 1).
+func (lim *Limiter) Allow() bool {
+	return lim.AllowN(time.Now(), 1)
+}
+
+// AllowN reports whether n events may happen at time now.
+// Use this method if you intend to drop / skip events that exceed the rate limit.
+// Otherwise use Reserve or Wait.
+func (lim *Limiter) AllowN(now time.Time, n int) bool {
+	return lim.reserveN(now, n, 0).ok
+}
+
+// A Reservation holds information about events that are permitted by a Limiter to happen after a delay.
+// A Reservation may be canceled, which may enable the Limiter to permit additional events.
+type Reservation struct {
+	ok        bool
+	lim       *Limiter
+	tokens    int
+	timeToAct time.Time
+	// This is the Limit at reservation time, it can change later.
+	limit Limit
+}
+
+// OK returns whether the limiter can provide the requested number of tokens
+// within the maximum wait time.  If OK is false, Delay returns InfDuration, and
+// Cancel does nothing.
+func (r *Reservation) OK() bool {
+	return r.ok
+}
+
+// Delay is shorthand for DelayFrom(time.Now()).
+func (r *Reservation) Delay() time.Duration {
+	return r.DelayFrom(time.Now())
+}
+
+// InfDuration is the duration returned by Delay when a Reservation is not OK.
+const InfDuration = time.Duration(1<<63 - 1)
+
+// DelayFrom returns the duration for which the reservation holder must wait
+// before taking the reserved action.  Zero duration means act immediately.
+// InfDuration means the limiter cannot grant the tokens requested in this
+// Reservation within the maximum wait time.
+func (r *Reservation) DelayFrom(now time.Time) time.Duration {
+	if !r.ok {
+		return InfDuration
+	}
+	delay := r.timeToAct.Sub(now)
+	if delay < 0 {
+		return 0
+	}
+	return delay
+}
+
+// Cancel is shorthand for CancelAt(time.Now()).
+func (r *Reservation) Cancel() {
+	r.CancelAt(time.Now())
+	return
+}
+
+// CancelAt indicates that the reservation holder will not perform the reserved action
+// and reverses the effects of this Reservation on the rate limit as much as possible,
+// considering that other reservations may have already been made.
+func (r *Reservation) CancelAt(now time.Time) {
+	if !r.ok {
+		return
+	}
+
+	r.lim.mu.Lock()
+	defer r.lim.mu.Unlock()
+
+	if r.lim.limit == Inf || r.tokens == 0 || r.timeToAct.Before(now) {
+		return
+	}
+
+	// calculate tokens to restore
+	// The duration between lim.lastEvent and r.timeToAct tells us how many tokens were reserved
+	// after r was obtained. These tokens should not be restored.
+	restoreTokens := float64(r.tokens) - r.limit.tokensFromDuration(r.lim.lastEvent.Sub(r.timeToAct))
+	if restoreTokens <= 0 {
+		return
+	}
+	// advance time to now
+	now, _, tokens := r.lim.advance(now)
+	// calculate new number of tokens
+	tokens += restoreTokens
+	if burst := float64(r.lim.burst); tokens > burst {
+		tokens = burst
+	}
+	// update state
+	r.lim.last = now
+	r.lim.tokens = tokens
+	if r.timeToAct == r.lim.lastEvent {
+		prevEvent := r.timeToAct.Add(r.limit.durationFromTokens(float64(-r.tokens)))
+		if !prevEvent.Before(now) {
+			r.lim.lastEvent = prevEvent
+		}
+	}
+
+	return
+}
+
+// Reserve is shorthand for ReserveN(time.Now(), 1).
+func (lim *Limiter) Reserve() *Reservation {
+	return lim.ReserveN(time.Now(), 1)
+}
+
+// ReserveN returns a Reservation that indicates how long the caller must wait before n events happen.
+// The Limiter takes this Reservation into account when allowing future events.
+// ReserveN returns false if n exceeds the Limiter's burst size.
+// Usage example:
+//   r := lim.ReserveN(time.Now(), 1)
+//   if !r.OK() {
+//     // Not allowed to act! Did you remember to set lim.burst to be > 0 ?
+//     return
+//   }
+//   time.Sleep(r.Delay())
+//   Act()
+// Use this method if you wish to wait and slow down in accordance with the rate limit without dropping events.
+// If you need to respect a deadline or cancel the delay, use Wait instead.
+// To drop or skip events exceeding rate limit, use Allow instead.
+func (lim *Limiter) ReserveN(now time.Time, n int) *Reservation {
+	r := lim.reserveN(now, n, InfDuration)
+	return &r
+}
+
+// Wait is shorthand for WaitN(ctx, 1).
+func (lim *Limiter) Wait(ctx context.Context) (err error) {
+	return lim.WaitN(ctx, 1)
+}
+
+// WaitN blocks until lim permits n events to happen.
+// It returns an error if n exceeds the Limiter's burst size, the Context is
+// canceled, or the expected wait time exceeds the Context's Deadline.
+// The burst limit is ignored if the rate limit is Inf.
+func (lim *Limiter) WaitN(ctx context.Context, n int) (err error) {
+	lim.mu.Lock()
+	burst := lim.burst
+	limit := lim.limit
+	lim.mu.Unlock()
+
+	if n > burst && limit != Inf {
+		return fmt.Errorf("rate: Wait(n=%d) exceeds limiter's burst %d", n, lim.burst)
+	}
+	// Check if ctx is already cancelled
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+	// Determine wait limit
+	now := time.Now()
+	waitLimit := InfDuration
+	if deadline, ok := ctx.Deadline(); ok {
+		waitLimit = deadline.Sub(now)
+	}
+	// Reserve
+	r := lim.reserveN(now, n, waitLimit)
+	if !r.ok {
+		return fmt.Errorf("rate: Wait(n=%d) would exceed context deadline", n)
+	}
+	// Wait if necessary
+	delay := r.DelayFrom(now)
+	if delay == 0 {
+		return nil
+	}
+	t := time.NewTimer(delay)
+	defer t.Stop()
+	select {
+	case <-t.C:
+		// We can proceed.
+		return nil
+	case <-ctx.Done():
+		// Context was canceled before we could proceed.  Cancel the
+		// reservation, which may permit other events to proceed sooner.
+		r.Cancel()
+		return ctx.Err()
+	}
+}
+
+// SetLimit is shorthand for SetLimitAt(time.Now(), newLimit).
+func (lim *Limiter) SetLimit(newLimit Limit) {
+	lim.SetLimitAt(time.Now(), newLimit)
+}
+
+// SetLimitAt sets a new Limit for the limiter. The new Limit, and Burst, may be violated
+// or underutilized by those which reserved (using Reserve or Wait) but did not yet act
+// before SetLimitAt was called.
+func (lim *Limiter) SetLimitAt(now time.Time, newLimit Limit) {
+	lim.mu.Lock()
+	defer lim.mu.Unlock()
+
+	now, _, tokens := lim.advance(now)
+
+	lim.last = now
+	lim.tokens = tokens
+	lim.limit = newLimit
+}
+
+// SetBurst is shorthand for SetBurstAt(time.Now(), newBurst).
+func (lim *Limiter) SetBurst(newBurst int) {
+	lim.SetBurstAt(time.Now(), newBurst)
+}
+
+// SetBurstAt sets a new burst size for the limiter.
+func (lim *Limiter) SetBurstAt(now time.Time, newBurst int) {
+	lim.mu.Lock()
+	defer lim.mu.Unlock()
+
+	now, _, tokens := lim.advance(now)
+
+	lim.last = now
+	lim.tokens = tokens
+	lim.burst = newBurst
+}
+
+// reserveN is a helper method for AllowN, ReserveN, and WaitN.
+// maxFutureReserve specifies the maximum reservation wait duration allowed.
+// reserveN returns Reservation, not *Reservation, to avoid allocation in AllowN and WaitN.
+func (lim *Limiter) reserveN(now time.Time, n int, maxFutureReserve time.Duration) Reservation {
+	lim.mu.Lock()
+
+	if lim.limit == Inf {
+		lim.mu.Unlock()
+		return Reservation{
+			ok:        true,
+			lim:       lim,
+			tokens:    n,
+			timeToAct: now,
+		}
+	}
+
+	now, last, tokens := lim.advance(now)
+
+	// Calculate the remaining number of tokens resulting from the request.
+	tokens -= float64(n)
+
+	// Calculate the wait duration
+	var waitDuration time.Duration
+	if tokens < 0 {
+		waitDuration = lim.limit.durationFromTokens(-tokens)
+	}
+
+	// Decide result
+	ok := n <= lim.burst && waitDuration <= maxFutureReserve
+
+	// Prepare reservation
+	r := Reservation{
+		ok:    ok,
+		lim:   lim,
+		limit: lim.limit,
+	}
+	if ok {
+		r.tokens = n
+		r.timeToAct = now.Add(waitDuration)
+	}
+
+	// Update state
+	if ok {
+		lim.last = now
+		lim.tokens = tokens
+		lim.lastEvent = r.timeToAct
+	} else {
+		lim.last = last
+	}
+
+	lim.mu.Unlock()
+	return r
+}
+
+// advance calculates and returns an updated state for lim resulting from the passage of time.
+// lim is not changed.
+func (lim *Limiter) advance(now time.Time) (newNow time.Time, newLast time.Time, newTokens float64) {
+	last := lim.last
+	if now.Before(last) {
+		last = now
+	}
+
+	// Avoid making delta overflow below when last is very old.
+	maxElapsed := lim.limit.durationFromTokens(float64(lim.burst) - lim.tokens)
+	elapsed := now.Sub(last)
+	if elapsed > maxElapsed {
+		elapsed = maxElapsed
+	}
+
+	// Calculate the new number of tokens, due to time that passed.
+	delta := lim.limit.tokensFromDuration(elapsed)
+	tokens := lim.tokens + delta
+	if burst := float64(lim.burst); tokens > burst {
+		tokens = burst
+	}
+
+	return now, last, tokens
+}
+
+// durationFromTokens is a unit conversion function from the number of tokens to the duration
+// of time it takes to accumulate them at a rate of limit tokens per second.
+func (limit Limit) durationFromTokens(tokens float64) time.Duration {
+	seconds := tokens / float64(limit)
+	return time.Nanosecond * time.Duration(1e9*seconds)
+}
+
+// tokensFromDuration is a unit conversion function from a time duration to the number of tokens
+// which could be accumulated during that duration at a rate of limit tokens per second.
+func (limit Limit) tokensFromDuration(d time.Duration) float64 {
+	// Split the integer and fractional parts ourself to minimize rounding errors.
+	// See golang.org/issues/34861.
+	sec := float64(d/time.Second) * float64(limit)
+	nsec := float64(d%time.Second) * float64(limit)
+	return sec + nsec/1e9
+}

vendor/modules.txt

@@ -83,3 +83,5 @@ golang.org/x/text/secure/bidirule
 golang.org/x/text/transform
 golang.org/x/text/unicode/bidi
 golang.org/x/text/unicode/norm
+# golang.org/x/time v0.0.0-20191024005414-555d28b269f0
+golang.org/x/time/rate