This is an automated email from the ASF dual-hosted git repository.
wwbmmm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new ae09e960 Serve builtin services only on ServerOptions.internal_port
(#3525)
ae09e960 is described below
commit ae09e960c7291605dda52356cc0c2d45567fb53e
Author: Bright Chen <[email protected]>
AuthorDate: Tue Sep 8 22:24:21 2026 +0800
Serve builtin services only on ServerOptions.internal_port (#3525)
---
docs/cn/server.md | 12 +-
docs/en/server.md | 12 +-
src/brpc/details/server_private_accessor.h | 50 +++++-
src/brpc/nshead_pb_service_adaptor.cpp | 17 ++-
src/brpc/policy/baidu_rpc_protocol.cpp | 23 ++-
src/brpc/policy/http_rpc_protocol.cpp | 6 +-
src/brpc/policy/hulu_pbrpc_protocol.cpp | 19 +--
src/brpc/policy/mongo_protocol.cpp | 3 +
src/brpc/policy/nshead_protocol.cpp | 3 +
src/brpc/policy/redis_protocol.cpp | 8 +
src/brpc/policy/sofa_pbrpc_protocol.cpp | 17 ++-
src/brpc/policy/thrift_protocol.cpp | 3 +
src/brpc/server.h | 3 +
test/brpc_http_rpc_protocol_unittest.cpp | 29 ++++
test/brpc_mongo_protocol_unittest.cpp | 23 +++
test/brpc_redis_unittest.cpp | 98 ++++++++++++
test/brpc_server_unittest.cpp | 235 +++++++++++++++++++++++++++++
test/brpc_socket_unittest.cpp | 23 ++-
18 files changed, 533 insertions(+), 51 deletions(-)
diff --git a/docs/cn/server.md b/docs/cn/server.md
index 7a9c47a4..565b7bb3 100644
--- a/docs/cn/server.md
+++ b/docs/cn/server.md
@@ -686,9 +686,19 @@ pthread模式可以让一些老代码快速尝试brpc，但我们仍然建议逐
-
设置内部端口。把ServerOptions.internal_port设为一个**仅允许内网访问**的端口。你可通过internal_port访问到内置服务，但通过对外端口(Server.Start时传入的那个)访问内置服务时将看到如下错误：
```
- [a27eda84bcdeef529a76f22872b78305] Not allowed to access builtin services,
try ServerOptions.internal_port=... instead if you're inside internal network
+ Not allowed to access builtin and Tabbed services, try
ServerOptions.internal_port=... instead if you're in internal network
```
+ 反过来，internal_port只提供内置服务(以及Tabbed服务)，普通服务的请求打到这个端口上会被拒绝：
+
+ ```
+ Only builtin and Tabbed services are accessible on
ServerOptions.internal_port=..., send the request to the port passed to
Server::Start() instead
+ ```
+
+
这是必须的：internal_port上的内置服务请求不需要通过ServerOptions.auth的鉴权，而鉴权结果是记在连接上的，一条连接只在第一个请求时鉴权一次。如果普通服务也在这个端口上提供，那么先发一个内置服务请求就能把整条连接标记为已鉴权，后续在同一条连接上访问普通服务将完全跳过鉴权。
+
+
有两类服务不走上面这条错误路径：ServerOptions.http_master_service和ServerOptions.baidu_master_service会接管所有URL和服务名(内置服务的也一样)，它们在internal_port上被直接忽略，请求回退到正常的服务查找，这样内置服务仍然可以从这个端口访问；ServerOptions.redis_service在解析阶段就把命令处理完了，没有Controller可以带回EPERM，因此internal_port上干脆不提供redis协议，连接会被直接关闭。
+
- http
proxy指定转发路径。nginx等可配置URL的映射关系，比如下面的配置把访问/MyAPI的外部流量映射到`target-server`的`/ServiceName/MethodName`。当外部流量尝试访问内置服务，比如说/status时，将直接被nginx拒绝。
```nginx
location /MyAPI {
diff --git a/docs/en/server.md b/docs/en/server.md
index 80806db5..f13193a8 100644
--- a/docs/en/server.md
+++ b/docs/en/server.md
@@ -680,9 +680,19 @@ Builtin services are useful, on the other hand include a
lot of internal informa
- Set internal port. Set ServerOptions.internal_port to a port which can
**only be accessible from internal**. You can view builtin services via
internal_port, while accesses from the public port (the one passed to
Server.Start) should see following error:
```
- [a27eda84bcdeef529a76f22872b78305] Not allowed to access builtin services,
try ServerOptions.internal_port=... instead if you're inside internal network
+ Not allowed to access builtin and Tabbed services, try
ServerOptions.internal_port=... instead if you're in internal network
```
+ Conversely internal_port serves builtin (and Tabbed) services only, requests
for ordinary services sent to it are rejected with:
+
+ ```
+ Only builtin and Tabbed services are accessible on
ServerOptions.internal_port=..., send the request to the port passed to
Server::Start() instead
+ ```
+
+ This is necessary: builtin requests on internal_port skip the authentication
of ServerOptions.auth, and the verdict is remembered per connection since a
connection is only authenticated once, on its first request. Were ordinary
services served there as well, sending a builtin request first would mark the
whole connection as authenticated and every later request on it would bypass
authentication entirely.
+
+ Two kinds of services do not take the error path above.
ServerOptions.http_master_service and ServerOptions.baidu_master_service answer
for every URL and service name, the builtin ones included, so they are ignored
on internal_port and the request falls back to the normal lookup, which keeps
the builtin services reachable from that port. ServerOptions.redis_service
handles the commands inside the parser and has no Controller to carry the
EPERM, so internal_port does not speak redis at [...]
+
- http proxies only proxy specified URLs. nginx etc is able to configure how
to map different URLs to back-end servers. For example the configure below maps
public traffic to /MyAPI to `/ServiceName/MethodName` of `target-server`. If
builtin services like /status are accessed from public, nginx rejects the
attempts directly.
```nginx
location /MyAPI {
diff --git a/src/brpc/details/server_private_accessor.h
b/src/brpc/details/server_private_accessor.h
index ee7929dd..0e6e4fbb 100644
--- a/src/brpc/details/server_private_accessor.h
+++ b/src/brpc/details/server_private_accessor.h
@@ -104,10 +104,23 @@ private:
const Server* _server;
};
+inline bool IsBuiltinOrTabbed(const Server::MethodProperty* mp) {
+ return nullptr != mp && (mp->is_builtin_service || mp->params.is_tabbed);
+}
+
+// True if `local_side' is ServerOptions.internal_port.
+inline bool IsInternalPort(const Server& server,
+ const butil::EndPoint& local_side) {
+ return server.options().internal_port >= 0 &&
+ local_side.port == server.options().internal_port;
+}
+
// Reject accesses to builtin services when the server is in security mode,
// in which case they are only reachable from ServerOptions.internal_port.
// Returns true if the access was rejected, in which case `cntl` was already
-// SetFailed() and the caller must stop dispatching the request immediately.
+// SetFailed() and the caller must not let the request reach the service's
+// normal processing. nshead hands the failed Controller to NsheadService
+// instead, as it does for the other pre-checks, see nshead_protocol.cpp.
// NOTE: Call this after ControllerPrivateAccessor::set_security_mode() and
// before the method is counted by MethodStatus::OnRequested(), so that
// rejected accesses do not pollute the stats of the method. `mp` may point
@@ -116,17 +129,42 @@ private:
// this beforehand, or make sure the listing is hidden in security mode.
inline bool RejectBuiltinAccess(Controller* cntl, const Server& server,
const Server::MethodProperty* mp) {
- if (!cntl->is_security_mode() ||
- (!mp->is_builtin_service && !mp->params.is_tabbed)) {
+ if (!cntl->is_security_mode() || !IsBuiltinOrTabbed(mp)) {
+ return false;
+ }
+ cntl->SetFailed(EPERM, "Not allowed to access builtin and Tabbed services,
try "
+ "ServerOptions.internal_port=%d instead if you're
in internal network",
+ server.options().internal_port);
+ return true;
+}
+
+// Reject accesses to non-builtin services arriving at
ServerOptions.internal_port,
+// which exposes builtin services away from the public listener, it is not a
second
+// entrance to the ordinary services of the server.
+// NOTE: Same return contract and placement rules as RejectBuiltinAccess().
+// This overload is for the protocols dispatching to a service that is never
+// builtin (NsheadService, ThriftService), hence has no MethodProperty.
+inline bool RejectNonBuiltinAccessFromInternalPort(Controller* cntl,
+ const Server& server) {
+ if (!IsInternalPort(server, cntl->local_side())) {
return false;
}
- cntl->SetFailed(EPERM, "Not allowed to access builtin services, try "
- "ServerOptions.internal_port=%d instead if
you're in "
- "internal network",
+ cntl->SetFailed(EPERM, "Only builtin and Tabbed services are accessible on
"
+ "ServerOptions.internal_port=%d, send the request
to the port "
+ "passed to Server::Start() instead",
server.options().internal_port);
return true;
}
+// Same as above, but lets builtin and Tabbed methods through.
+inline bool RejectNonBuiltinAccessFromInternalPort(
+ Controller* cntl, const Server& server, const Server::MethodProperty* mp) {
+ if (IsBuiltinOrTabbed(mp)) {
+ return false;
+ }
+ return RejectNonBuiltinAccessFromInternalPort(cntl, server);
+}
+
// Count one error if release() is not called before destruction of this
object.
class ScopedNonServiceError {
public:
diff --git a/src/brpc/nshead_pb_service_adaptor.cpp
b/src/brpc/nshead_pb_service_adaptor.cpp
index 17e29228..3be93224 100644
--- a/src/brpc/nshead_pb_service_adaptor.cpp
+++ b/src/brpc/nshead_pb_service_adaptor.cpp
@@ -114,22 +114,23 @@ void NsheadPbServiceAdaptor::ProcessNsheadRequest(
}
ServerPrivateAccessor server_accessor(&server);
- const Server::MethodProperty *sp = server_accessor
+ const Server::MethodProperty* mp = server_accessor
.FindMethodPropertyByFullName(meta->full_method_name());
- if (nullptr == sp ||
- sp->service->GetDescriptor() == BadMethodService::descriptor()) {
+ if (nullptr == mp ||
+ mp->service->GetDescriptor() == BadMethodService::descriptor()) {
controller->SetFailed(ENOMETHOD, "Fail to find method=%s",
meta->full_method_name().c_str());
break;
}
- if (RejectBuiltinAccess(controller, server, sp)) {
+ if (RejectBuiltinAccess(controller, server, mp) ||
+ RejectNonBuiltinAccessFromInternalPort(controller, server, mp)) {
break;
}
- pbdone->status = sp->status;
- sp->status->OnRequested();
+ pbdone->status = mp->status;
+ mp->status->OnRequested();
- google::protobuf::Service* svc = sp->service;
- const google::protobuf::MethodDescriptor* method = sp->method;
+ google::protobuf::Service* svc = mp->service;
+ const google::protobuf::MethodDescriptor* method = mp->method;
ControllerPrivateAccessor(controller).set_method(method);
done->SetMethodName(butil::EnsureString(method->full_name()));
pbdone->pbreq.reset(svc->GetRequestPrototype(method).New());
diff --git a/src/brpc/policy/baidu_rpc_protocol.cpp
b/src/brpc/policy/baidu_rpc_protocol.cpp
index 5a6451b0..baa691e0 100644
--- a/src/brpc/policy/baidu_rpc_protocol.cpp
+++ b/src/brpc/policy/baidu_rpc_protocol.cpp
@@ -278,6 +278,12 @@ struct BaiduProxyPBMessages : public RpcPBMessages {
};
}
+static bool IsBaiduMasterService(const Server* server,
+ const butil::EndPoint& local_side) {
+ return nullptr != server->options().baidu_master_service &&
+ !IsInternalPort(*server, local_side);
+}
+
// Used by UT, can't be static.
void SendRpcResponse(int64_t correlation_id, Controller* cntl,
RpcPBMessages* messages, const Server* server,
@@ -306,10 +312,10 @@ void SendRpcResponse(int64_t correlation_id, Controller*
cntl,
}
cntl->CallAfterRpcResp(req, res);
- if (nullptr == server->options().baidu_master_service) {
- server->options().rpc_pb_message_factory->Return(messages);
- } else {
+ if (IsBaiduMasterService(server, cntl->local_side())) {
BaiduProxyPBMessages::Return(static_cast<BaiduProxyPBMessages*>(messages));
+ } else {
+ server->options().rpc_pb_message_factory->Return(messages);
}
};
@@ -713,13 +719,13 @@ void ProcessRpcRequest(InputMessageBase* msg_base) {
google::protobuf::Service* svc = nullptr;
google::protobuf::MethodDescriptor* method = nullptr;
- if (nullptr != server->options().baidu_master_service) {
+ if (IsBaiduMasterService(server, cntl->local_side())) {
if (socket->is_overcrowded() &&
!server->options().ignore_eovercrowded &&
!server->options().baidu_master_service->ignore_eovercrowded()) {
- cntl->SetFailed(EOVERCROWDED, "Connection to %s is
overcrowded",
-
butil::endpoint2str(socket->remote_side()).c_str());
- break;
+ cntl->SetFailed(EOVERCROWDED, "Connection to %s is
overcrowded",
+
butil::endpoint2str(socket->remote_side()).c_str());
+ break;
}
svc = server->options().baidu_master_service;
auto sampled_request = new SampledRequest;
@@ -770,7 +776,8 @@ void ProcessRpcRequest(InputMessageBase* msg_base) {
request_meta.method_name().c_str());
break;
}
- if (RejectBuiltinAccess(cntl.get(), *server, mp)) {
+ if (RejectBuiltinAccess(cntl.get(), *server, mp) ||
+ RejectNonBuiltinAccessFromInternalPort(cntl.get(), *server,
mp)) {
break;
}
if (mp->service->GetDescriptor() ==
BadMethodService::descriptor()) {
diff --git a/src/brpc/policy/http_rpc_protocol.cpp
b/src/brpc/policy/http_rpc_protocol.cpp
index 4d1c8824..eb2ac11b 100644
--- a/src/brpc/policy/http_rpc_protocol.cpp
+++ b/src/brpc/policy/http_rpc_protocol.cpp
@@ -1558,7 +1558,8 @@ void ProcessHttpRequest(InputMessageBase *msg) {
return;
}
- if (server->options().http_master_service) {
+ if (server->options().http_master_service &&
+ !IsInternalPort(*server, cntl->local_side())) {
// If http_master_service is on, just call it.
google::protobuf::Service* svc = server->options().http_master_service;
const google::protobuf::MethodDescriptor* md =
@@ -1605,7 +1606,8 @@ void ProcessHttpRequest(InputMessageBase *msg) {
mp->service->CallMethod(mp->method, cntl, &breq, &bres, nullptr);
return;
}
- if (RejectBuiltinAccess(cntl, *server, mp)) {
+ if (RejectBuiltinAccess(cntl, *server, mp) ||
+ RejectNonBuiltinAccessFromInternalPort(cntl, *server, mp)) {
return;
}
// Switch to service-specific error.
diff --git a/src/brpc/policy/hulu_pbrpc_protocol.cpp
b/src/brpc/policy/hulu_pbrpc_protocol.cpp
index 4bacd9e5..146f8858 100644
--- a/src/brpc/policy/hulu_pbrpc_protocol.cpp
+++ b/src/brpc/policy/hulu_pbrpc_protocol.cpp
@@ -442,27 +442,28 @@ void ProcessHuluRequest(InputMessageBase* msg_base) {
break;
}
- const Server::MethodProperty *sp =
+ const Server::MethodProperty* mp =
server_accessor.FindMethodPropertyByNameAndIndex(
meta.service_name(), meta.method_index());
- if (nullptr == sp) {
+ if (nullptr == mp) {
cntl->SetFailed(ENOMETHOD, "Fail to find method=%d of service=%s",
meta.method_index(), meta.service_name().c_str());
break;
}
- if (RejectBuiltinAccess(cntl.get(), *server, sp)) {
+ if (RejectBuiltinAccess(cntl.get(), *server, mp) ||
+ RejectNonBuiltinAccessFromInternalPort(cntl.get(), *server, mp)) {
break;
}
- if (sp->service->GetDescriptor() == BadMethodService::descriptor()) {
+ if (mp->service->GetDescriptor() == BadMethodService::descriptor()) {
BadMethodRequest breq;
BadMethodResponse bres;
breq.set_service_name(meta.service_name());
- sp->service->CallMethod(sp->method, cntl.get(), &breq, &bres,
nullptr);
+ mp->service->CallMethod(mp->method, cntl.get(), &breq, &bres,
nullptr);
break;
}
if (socket->is_overcrowded() &&
!server->options().ignore_eovercrowded &&
- !sp->ignore_eovercrowded) {
+ !mp->ignore_eovercrowded) {
cntl->SetFailed(EOVERCROWDED, "Connection to %s is overcrowded",
butil::endpoint2str(socket->remote_side()).c_str());
break;
@@ -470,8 +471,8 @@ void ProcessHuluRequest(InputMessageBase* msg_base) {
// Switch to service-specific error.
non_service_error.release();
- method_status = sp->status;
- const google::protobuf::MethodDescriptor* method = sp->method;
+ method_status = mp->status;
+ const google::protobuf::MethodDescriptor* method = mp->method;
const std::string method_full_name =
butil::EnsureString(method->full_name());
if (method_status) {
int rejected_cc = 0;
@@ -482,7 +483,7 @@ void ProcessHuluRequest(InputMessageBase* msg_base) {
}
}
- google::protobuf::Service* svc = sp->service;
+ google::protobuf::Service* svc = mp->service;
accessor.set_method(method);
if (!server->AcceptRequest(cntl.get())) {
diff --git a/src/brpc/policy/mongo_protocol.cpp
b/src/brpc/policy/mongo_protocol.cpp
index 3c7262d4..35647d79 100644
--- a/src/brpc/policy/mongo_protocol.cpp
+++ b/src/brpc/policy/mongo_protocol.cpp
@@ -252,6 +252,9 @@ void ProcessMongoRequest(InputMessageBase* msg_base) {
mongo_done->cntl.SetFailed(ENOMETHOD, "Fail to find
default_method");
break;
}
+ if (RejectNonBuiltinAccessFromInternalPort(&mongo_done->cntl, *server,
mp)) {
+ break;
+ }
// Switch to service-specific error.
non_service_error.release();
MethodStatus* method_status = mp->status;
diff --git a/src/brpc/policy/nshead_protocol.cpp
b/src/brpc/policy/nshead_protocol.cpp
index 72fc8994..5042b922 100644
--- a/src/brpc/policy/nshead_protocol.cpp
+++ b/src/brpc/policy/nshead_protocol.cpp
@@ -314,6 +314,9 @@ void ProcessNsheadRequest(InputMessageBase* msg_base) {
cntl->SetFailed(ELOGOFF, "Server is stopping");
break;
}
+ if (RejectNonBuiltinAccessFromInternalPort(cntl, *server)) {
+ break;
+ }
if (socket->is_overcrowded() &&
!server->options().ignore_eovercrowded) {
cntl->SetFailed(EOVERCROWDED, "Connection to %s is overcrowded",
butil::endpoint2str(socket->remote_side()).c_str());
diff --git a/src/brpc/policy/redis_protocol.cpp
b/src/brpc/policy/redis_protocol.cpp
index 3009d7b7..e46bed95 100644
--- a/src/brpc/policy/redis_protocol.cpp
+++ b/src/brpc/policy/redis_protocol.cpp
@@ -125,6 +125,14 @@ ParseResult ParseRedisMessage(butil::IOBuf* source,
Socket* socket,
if (!rs) {
return MakeParseError(PARSE_ERROR_TRY_OTHERS);
}
+ if (IsInternalPort(*server, socket->local_side())) {
+ // ServerOptions.internal_port serves builtin and Tabbed services
+ // only and a RedisService is neither. The command handlers run
+ // right here rather than in ProcessRedisRequest(), there is no
+ // Controller to reject the request with, so behave as if redis
+ // was not enabled on this port at all.
+ return MakeParseError(PARSE_ERROR_TRY_OTHERS);
+ }
RedisConnContext* ctx =
static_cast<RedisConnContext*>(socket->parsing_context());
if (ctx == nullptr) {
ctx = new RedisConnContext(rs);
diff --git a/src/brpc/policy/sofa_pbrpc_protocol.cpp
b/src/brpc/policy/sofa_pbrpc_protocol.cpp
index d0c42cc4..fa512590 100644
--- a/src/brpc/policy/sofa_pbrpc_protocol.cpp
+++ b/src/brpc/policy/sofa_pbrpc_protocol.cpp
@@ -403,36 +403,37 @@ void ProcessSofaRequest(InputMessageBase* msg_base) {
break;
}
- const Server::MethodProperty *sp =
+ const Server::MethodProperty* mp =
server_accessor.FindMethodPropertyByFullName(meta.method());
- if (nullptr == sp) {
+ if (nullptr == mp) {
cntl->SetFailed(ENOMETHOD, "Fail to find method=%s",
meta.method().c_str());
break;
}
- if (RejectBuiltinAccess(cntl.get(), *server, sp)) {
+ if (RejectBuiltinAccess(cntl.get(), *server, mp) ||
+ RejectNonBuiltinAccessFromInternalPort(cntl.get(), *server, mp)) {
break;
}
if (socket->is_overcrowded() &&
!server->options().ignore_eovercrowded &&
- !sp->ignore_eovercrowded) {
+ !mp->ignore_eovercrowded) {
cntl->SetFailed(EOVERCROWDED, "Connection to %s is overcrowded",
butil::endpoint2str(socket->remote_side()).c_str());
break;
}
// Switch to service-specific error.
non_service_error.release();
- method_status = sp->status;
+ method_status = mp->status;
if (method_status) {
int rejected_cc = 0;
if (!method_status->OnRequested(&rejected_cc)) {
cntl->SetFailed(ELIMIT, "Rejected by %s's ConcurrencyLimiter,
concurrency=%d",
-
butil::EnsureString(sp->method->full_name()).c_str(), rejected_cc);
+
butil::EnsureString(mp->method->full_name()).c_str(), rejected_cc);
break;
}
}
- google::protobuf::Service* svc = sp->service;
- const google::protobuf::MethodDescriptor* method = sp->method;
+ google::protobuf::Service* svc = mp->service;
+ const google::protobuf::MethodDescriptor* method = mp->method;
accessor.set_method(method);
if (!server->AcceptRequest(cntl.get())) {
diff --git a/src/brpc/policy/thrift_protocol.cpp
b/src/brpc/policy/thrift_protocol.cpp
index e97dd00b..735abde4 100755
--- a/src/brpc/policy/thrift_protocol.cpp
+++ b/src/brpc/policy/thrift_protocol.cpp
@@ -510,6 +510,9 @@ void ProcessThriftRequest(InputMessageBase* msg_base) {
" ServerOptions.thrift_service, close the connection.";
return cntl->SetFailed(EINTERNAL, "ServerOptions.thrift_service is
NULL");
}
+ if (RejectNonBuiltinAccessFromInternalPort(cntl, *server)) {
+ return;
+ }
// Switch to service-specific error.
non_service_error.release();
diff --git a/src/brpc/server.h b/src/brpc/server.h
index 276e8795..6e7d2b2b 100644
--- a/src/brpc/server.h
+++ b/src/brpc/server.h
@@ -202,6 +202,9 @@ struct ServerOptions {
// hiding them from public. Setting this option also enables security
// protection code which we may add constantly.
// Update: this option affects Tabbed services as well.
+ // Update: this port carries builtin and Tabbed services only, requests
+ // for ordinary services are rejected with EPERM and must be sent to the
+ // port passed to Start().
// Default: -1
int internal_port;
diff --git a/test/brpc_http_rpc_protocol_unittest.cpp
b/test/brpc_http_rpc_protocol_unittest.cpp
index f23bbfb7..10fb5cc3 100644
--- a/test/brpc_http_rpc_protocol_unittest.cpp
+++ b/test/brpc_http_rpc_protocol_unittest.cpp
@@ -563,6 +563,35 @@ TEST_F(HttpTest,
builtin_auth_policy_on_public_and_internal_port) {
ASSERT_TRUE(protected_cntl.Failed());
}
+ {
+ // A builtin request is exempted from authentication on internal_port
+ // and its verdict latches the whole connection, so the exemption would
+ // carry over to whatever is sent next on that very connection. Only
+ // builtin services are served there, which keeps the latch harmless.
+ const std::string connection_group = "builtin-auth-policy-internal";
+ brpc::Channel builtin_channel;
+ brpc::Channel protected_channel;
+ brpc::ChannelOptions copt;
+ copt.protocol = brpc::PROTOCOL_HTTP;
+ copt.connection_type = brpc::CONNECTION_TYPE_POOLED;
+ copt.connection_group = connection_group;
+ copt.max_retry = 0;
+ ASSERT_EQ(0, builtin_channel.Init(internal_ep, &copt));
+ ASSERT_EQ(0, protected_channel.Init(internal_ep, &copt));
+
+ brpc::Controller builtin_cntl;
+ CallVersion(&builtin_channel, &builtin_cntl);
+ ASSERT_FALSE(builtin_cntl.Failed()) << builtin_cntl.ErrorText();
+ ASSERT_EQ(brpc::HTTP_STATUS_OK,
builtin_cntl.http_response().status_code());
+
+ brpc::Controller protected_cntl;
+ CallHttpEcho(&protected_channel, &protected_cntl);
+ ASSERT_TRUE(protected_cntl.Failed());
+ ASSERT_EQ(brpc::EHTTP, protected_cntl.ErrorCode()) <<
protected_cntl.ErrorText();
+ ASSERT_EQ(brpc::HTTP_STATUS_FORBIDDEN,
+ protected_cntl.http_response().status_code());
+ }
+
ASSERT_EQ(0, server.Stop(0));
ASSERT_EQ(0, server.Join());
brpc::FLAGS_max_connection_pool_size = saved_max_connection_pool_size;
diff --git a/test/brpc_mongo_protocol_unittest.cpp
b/test/brpc_mongo_protocol_unittest.cpp
index 68e4acdc..3ec22566 100644
--- a/test/brpc_mongo_protocol_unittest.cpp
+++ b/test/brpc_mongo_protocol_unittest.cpp
@@ -181,6 +181,29 @@ TEST_F(MongoTest, process_request_failed_socket) {
ASSERT_EQ(0ll, _server._nerror_bvar.get_value());
}
+TEST_F(MongoTest, process_request_from_internal_port) {
+ // The fixture talks over a pipe, which has no local address at all while
+ // Server::Start() never accepts 0 as ServerOptions.internal_port. Give the
+ // connection an address the server can recognize as internal, the port is
+ // arbitrary as nothing listens on it.
+ _socket->_local_side = butil::EndPoint(butil::IP_ANY, 8888);
+ _server._options.internal_port = _socket->local_side().port;
+
+ brpc::mongo_head_t header = { 0, 0, 0, 0 };
+ header.op_code = brpc::MONGO_OPCODE_REPLY;
+ header.message_length = sizeof(header) + EXP_REQUEST.length();
+ butil::IOBuf total_buf;
+ total_buf.append(static_cast<const void*>(&header), sizeof(header));
+ total_buf.append(EXP_REQUEST);
+ brpc::ParseResult req_pr = brpc::policy::ParseMongoMessage(
+ &total_buf, _socket.get(), false, &_server);
+ ASSERT_EQ(brpc::PARSE_OK, req_pr.error());
+ ProcessMessage(brpc::policy::ProcessMongoRequest, req_pr.message(), false);
+ // MyEchoService was never reached, the request was counted as an error of
+ // the server rather than as a call of the method.
+ ASSERT_EQ(1ll, _server._nerror_bvar.get_value());
+}
+
TEST_F(MongoTest, complete_flow) {
butil::IOBuf request_buf;
butil::IOBuf total_buf;
diff --git a/test/brpc_redis_unittest.cpp b/test/brpc_redis_unittest.cpp
index f4830d04..724b5e77 100644
--- a/test/brpc_redis_unittest.cpp
+++ b/test/brpc_redis_unittest.cpp
@@ -19,6 +19,8 @@
#include <iostream>
#include <memory>
#include <unordered_map>
+#include <butil/endpoint.h>
+#include <butil/fd_guard.h>
#include <butil/time.h>
#include <butil/logging.h>
#include <brpc/redis.h>
@@ -1223,6 +1225,102 @@ TEST_F(RedisTest, server_sanity) {
ASSERT_EQ("", response.reply(3).data());
}
+// Returns a port nothing is listening on, or -1. ServerOptions.internal_port
+// has to be an explicit number, Server::Start() rejects 0 because it stands
+// for an ephemeral port, so ask the system for a free one rather than hardcode
+// a port that another test may be listening on.
+static int PickUnusedPort() {
+ butil::fd_guard sockfd(butil::tcp_listen(butil::EndPoint(butil::IP_ANY,
0)));
+ if (sockfd < 0) {
+ return -1;
+ }
+ butil::EndPoint point;
+ if (butil::get_local_side(sockfd, &point) != 0) {
+ return -1;
+ }
+ return point.port;
+}
+
+// Starts `server' on an ephemeral port and fills options->internal_port with
+// another one. Both are released before Start() binds them and something else
+// may take one in between, hence the retries. Returns 0 on success.
+static int StartWithInternalPort(brpc::Server* server,
+ brpc::ServerOptions* options) {
+ for (int i = 0; i < 10; ++i) {
+ int internal_port = PickUnusedPort();
+ if (internal_port < 0) {
+ continue;
+ }
+ options->internal_port = internal_port;
+ if (0 == server->Start("127.0.0.1:0", options)) {
+ return 0;
+ }
+ }
+ return -1;
+}
+
+TEST_F(RedisTest, server_is_not_served_on_internal_port) {
+ std::string password = GeneratePassword();
+ std::unique_ptr<brpc::policy::RedisAuthenticator> redis_auth_holder(
+ new brpc::policy::RedisAuthenticator(password));
+ RedisServiceImpl* rsimpl = new RedisServiceImpl(password);
+ std::unique_ptr<SetCommandHandler> sh(new SetCommandHandler(rsimpl));
+ std::unique_ptr<AuthCommandHandler> ah(new AuthCommandHandler(rsimpl));
+ rsimpl->AddCommandHandler("set", sh.get());
+ rsimpl->AddCommandHandler("auth", ah.get());
+
+ brpc::Server server;
+ brpc::ServerOptions server_options;
+ server_options.redis_service = rsimpl;
+ ASSERT_EQ(0, StartWithInternalPort(&server, &server_options));
+
+ brpc::ChannelOptions options;
+ options.protocol = brpc::PROTOCOL_REDIS;
+ options.auth = redis_auth_holder.get();
+ options.max_retry = 0;
+
+ brpc::RedisRequest request;
+ ASSERT_TRUE(request.AddCommand("set key1 value1"));
+
+ // The internal port is up and serving its builtin services, it just does
+ // not speak redis, so the command below fails to be parsed rather than
+ // fails to be sent.
+ brpc::ChannelOptions http_options;
+ http_options.protocol = brpc::PROTOCOL_HTTP;
+ http_options.max_retry = 0;
+ brpc::Channel http_channel;
+ ASSERT_EQ(0, http_channel.Init(
+ "127.0.0.1", server_options.internal_port, &http_options));
+ brpc::Controller http_cntl;
+ http_cntl.http_request().uri() = "/version";
+ http_channel.CallMethod(nullptr, &http_cntl, nullptr, nullptr, nullptr);
+ ASSERT_FALSE(http_cntl.Failed()) << http_cntl.ErrorText();
+
+ brpc::Channel internal_channel;
+ ASSERT_EQ(0, internal_channel.Init(
+ "127.0.0.1", server_options.internal_port, &options));
+ brpc::RedisResponse response;
+ brpc::Controller cntl;
+ internal_channel.CallMethod(nullptr, &cntl, &request, &response, nullptr);
+ ASSERT_TRUE(cntl.Failed());
+ ASSERT_EQ(0, response.reply_size());
+
+ // The port passed to Start() speaks redis as before.
+ brpc::Channel channel;
+ ASSERT_EQ(0, channel.Init(
+ "127.0.0.1", server.listen_address().port, &options));
+ cntl.Reset();
+ response.Clear();
+ channel.CallMethod(nullptr, &cntl, &request, &response, nullptr);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ ASSERT_EQ(1, response.reply_size());
+ ASSERT_EQ(brpc::REDIS_REPLY_STATUS, response.reply(0).type());
+ ASSERT_STREQ("OK", response.reply(0).c_str());
+
+ ASSERT_EQ(0, server.Stop(0));
+ ASSERT_EQ(0, server.Join());
+}
+
void* incr_thread(void* arg) {
brpc::Channel* c = static_cast<brpc::Channel*>(arg);
for (int i = 0; i < 5000; ++i) {
diff --git a/test/brpc_server_unittest.cpp b/test/brpc_server_unittest.cpp
index ed1fa91c..e94d8508 100644
--- a/test/brpc_server_unittest.cpp
+++ b/test/brpc_server_unittest.cpp
@@ -48,6 +48,7 @@
#include "brpc/builtin/sockets_service.h" // SocketsService
#include "brpc/builtin/bad_method_service.h"
#include "brpc/server.h"
+#include "brpc/nshead_service.h"
#include "brpc/restful.h"
#include "brpc/channel.h"
#include "brpc/socket_map.h"
@@ -57,6 +58,7 @@
#include "v1.pb.h"
#include "v2.pb.h"
#include "v3.pb.h"
+#include "health_check.pb.h"
int main(int argc, char* argv[]) {
testing::InitGoogleTest(&argc, argv);
@@ -1719,6 +1721,158 @@ TEST_F(ServerTest,
builtin_services_are_gated_by_internal_port) {
ASSERT_EQ(0, server.Join());
}
+// Call the same ordinary service the way a browser would.
+void CallEchoByHttp(const butil::EndPoint& ep, brpc::Controller* cntl) {
+ brpc::ChannelOptions copt;
+ copt.protocol = brpc::PROTOCOL_HTTP;
+ copt.max_retry = 0;
+ brpc::Channel chan;
+ ASSERT_EQ(0, chan.Init(ep, &copt));
+ test::EchoRequest req;
+ test::EchoResponse res;
+ req.set_message(EXP_REQUEST);
+ cntl->http_request().uri() = "/EchoService/Echo";
+ cntl->http_request().set_method(brpc::HTTP_METHOD_POST);
+ cntl->http_request().set_content_type("application/json");
+ chan.CallMethod(nullptr, cntl, &req, &res, nullptr);
+}
+
+// Returns a port nothing is listening on, or -1. `ServerOptions.internal_port`
+// has to be an explicit number, Server::Start() rejects 0 because it stands
+// for an ephemeral port, so ask the system for a free one rather than hardcode
+// a port that another test may be listening on.
+int PickUnusedPort() {
+ butil::fd_guard sockfd(butil::tcp_listen(butil::EndPoint(butil::IP_ANY,
0)));
+ if (sockfd < 0) {
+ return -1;
+ }
+ butil::EndPoint point;
+ if (butil::get_local_side(sockfd, &point) != 0) {
+ return -1;
+ }
+ return point.port;
+}
+
+// Starts `server` on an ephemeral port and fills `options->internal_port` with
+// another one. Both are released before Start() binds them and something else
+// may take one in between, hence the retries. Returns 0 on success.
+int StartWithInternalPort(brpc::Server* server, brpc::ServerOptions* options) {
+ for (int i = 0; i < 10; ++i) {
+ int internal_port = PickUnusedPort();
+ if (internal_port < 0) {
+ continue;
+ }
+ options->internal_port = internal_port;
+ if (0 == server->Start("127.0.0.1:0", options)) {
+ return 0;
+ }
+ }
+ return -1;
+}
+
+TEST_F(ServerTest, ordinary_services_are_not_served_on_internal_port) {
+ const struct {
+ brpc::ProtocolType protocol;
+ const char* name;
+ } cases[] = {
+ { brpc::PROTOCOL_BAIDU_STD, "baidu_std" },
+ { brpc::PROTOCOL_HULU_PBRPC, "hulu_pbrpc" },
+ { brpc::PROTOCOL_SOFA_PBRPC, "sofa_pbrpc" },
+ };
+
+ brpc::Server server;
+ EchoServiceImpl echo_svc;
+ ASSERT_EQ(0, server.AddService(&echo_svc,
brpc::SERVER_DOESNT_OWN_SERVICE));
+ brpc::ServerOptions opt;
+ ASSERT_EQ(0, StartWithInternalPort(&server, &opt));
+ butil::EndPoint ep = server.listen_address();
+ butil::EndPoint internal_ep(ep.ip, opt.internal_port);
+
+ for (size_t i = 0; i < arraysize(cases); ++i) {
+ brpc::Controller cntl;
+ CallEchoByPb(internal_ep, cases[i].protocol, &cntl);
+ ASSERT_EQ(EPERM, cntl.ErrorCode())
+ << cases[i].name << ": " << cntl.ErrorText();
+
+ // The public port is where ordinary services live.
+ cntl.Reset();
+ CallEchoByPb(ep, cases[i].protocol, &cntl);
+ ASSERT_FALSE(cntl.Failed())
+ << cases[i].name << ": " << cntl.ErrorText();
+ }
+
+ brpc::Controller cntl;
+ CallEchoByHttp(internal_ep, &cntl);
+ ASSERT_TRUE(cntl.Failed());
+ ASSERT_EQ(brpc::HTTP_STATUS_FORBIDDEN, cntl.http_response().status_code())
+ << cntl.ErrorText();
+ cntl.Reset();
+ CallEchoByHttp(ep, &cntl);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+
+ ASSERT_EQ(0, server.Stop(0));
+ ASSERT_EQ(0, server.Join());
+}
+
+// NsheadService is dispatched to without a MethodProperty, the gate has to be
+// applied by the protocol itself. The service echoes back what the framework
+// decided so that the client can tell an acceptance from a rejection: nshead
+// carries no error field.
+class EchoNsheadService : public brpc::NsheadService {
+public:
+ void ProcessNsheadRequest(const brpc::Server&,
+ brpc::Controller* cntl,
+ const brpc::NsheadMessage& request,
+ brpc::NsheadMessage* response,
+ brpc::NsheadClosure* done) override {
+ brpc::ClosureGuard done_guard(done);
+ if (cntl->Failed()) {
+ response->body.append(butil::string_printf("%d",
cntl->ErrorCode()));
+ return;
+ }
+ response->body.append(EXP_RESPONSE);
+ }
+};
+
+// Same gate as ordinary_services_are_not_served_on_internal_port, for the
+// protocols that dispatch to a service which is never builtin. Their verify()
+// refuses every request when ServerOptions.auth is set, so the connection
+// latched by an exempted builtin request is the only way to reach them.
+TEST_F(ServerTest, nshead_service_is_not_served_on_internal_port) {
+ brpc::Server server;
+ brpc::ServerOptions opt;
+ opt.nshead_service = new EchoNsheadService;
+ ASSERT_EQ(0, StartWithInternalPort(&server, &opt));
+ butil::EndPoint ep = server.listen_address();
+ butil::EndPoint internal_ep(ep.ip, opt.internal_port);
+
+ brpc::ChannelOptions copt;
+ copt.protocol = brpc::PROTOCOL_NSHEAD;
+ copt.connection_type = brpc::CONNECTION_TYPE_POOLED;
+ copt.max_retry = 0;
+
+ brpc::Channel internal_chan;
+ ASSERT_EQ(0, internal_chan.Init(internal_ep, &copt));
+ brpc::NsheadMessage req;
+ brpc::NsheadMessage res;
+ brpc::Controller cntl;
+ req.body.append(EXP_REQUEST);
+ internal_chan.CallMethod(nullptr, &cntl, &req, &res, nullptr);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ ASSERT_EQ(butil::string_printf("%d", EPERM), res.body.to_string());
+
+ brpc::Channel chan;
+ ASSERT_EQ(0, chan.Init(ep, &copt));
+ cntl.Reset();
+ res.body.clear();
+ chan.CallMethod(nullptr, &cntl, &req, &res, nullptr);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ ASSERT_EQ(EXP_RESPONSE, res.body.to_string());
+
+ ASSERT_EQ(0, server.Stop(0));
+ ASSERT_EQ(0, server.Join());
+}
+
// A service-name-only URL is dispatched to the builtin BadMethodService which
// lists the methods of the service.
void CallServiceWithoutMethodByHttp(const butil::EndPoint& ep,
@@ -2144,6 +2298,87 @@ TEST_F(ServerTest, baidu_master_service) {
ASSERT_EQ(0, server.Join());
}
+class HttpMasterServiceImpl : public test::HealthCheckTestService {
+public:
+ void default_method(google::protobuf::RpcController* cntl_base,
+ const test::HealthCheckRequest*,
+ test::HealthCheckResponse*,
+ google::protobuf::Closure* done) override {
+ brpc::ClosureGuard done_guard(done);
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
+ cntl->response_attachment().append(EXP_RESPONSE);
+ }
+};
+
+// http_master_service answers every URL and carries the payload in the
+// attachments rather than in pb messages.
+void CallEchoByHttpWithoutPb(const butil::EndPoint& ep, brpc::Controller*
cntl) {
+ brpc::ChannelOptions copt;
+ copt.protocol = brpc::PROTOCOL_HTTP;
+ copt.max_retry = 0;
+ brpc::Channel chan;
+ ASSERT_EQ(0, chan.Init(ep, &copt));
+ cntl->http_request().uri() = "/EchoService/Echo";
+ chan.CallMethod(nullptr, cntl, nullptr, nullptr, nullptr);
+}
+
+TEST_F(ServerTest, master_services_are_not_served_on_internal_port) {
+ brpc::Server server;
+ EchoServiceImpl echo_svc;
+ ASSERT_EQ(0, server.AddService(&echo_svc,
brpc::SERVER_DOESNT_OWN_SERVICE));
+ brpc::ServerOptions opt;
+ opt.baidu_master_service = new BaiduMasterServiceImpl;
+ opt.http_master_service = new HttpMasterServiceImpl;
+ ASSERT_EQ(0, StartWithInternalPort(&server, &opt));
+ butil::EndPoint ep = server.listen_address();
+ butil::EndPoint internal_ep(ep.ip, opt.internal_port);
+
+ // The master services answer on the public port.
+ brpc::ChannelOptions copt;
+ copt.protocol = brpc::PROTOCOL_BAIDU_STD;
+ copt.max_retry = 0;
+ brpc::Channel channel;
+ ASSERT_EQ(0, channel.Init(ep, &copt));
+ TestBaiduMasterService(channel, brpc::COMPRESS_TYPE_NONE);
+
+ brpc::Controller cntl;
+ CallEchoByHttpWithoutPb(ep, &cntl);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ ASSERT_EQ(EXP_RESPONSE, cntl.response_attachment().to_string());
+
+ // Not on internal_port though, where the request falls back to the normal
+ // resolution and EchoService is rejected as any other ordinary service.
+ cntl.Reset();
+ CallEchoByPb(internal_ep, brpc::PROTOCOL_BAIDU_STD, &cntl);
+ ASSERT_EQ(EPERM, cntl.ErrorCode()) << cntl.ErrorText();
+
+ cntl.Reset();
+ CallEchoByHttpWithoutPb(internal_ep, &cntl);
+ ASSERT_TRUE(cntl.Failed());
+ ASSERT_EQ(brpc::HTTP_STATUS_FORBIDDEN, cntl.http_response().status_code())
+ << cntl.ErrorText();
+
+ // The builtin services are still served there, which is what the port is
+ // for. Rejecting the master service outright would have hidden them.
+ cntl.Reset();
+ CallVersionByPb(internal_ep, brpc::PROTOCOL_BAIDU_STD, &cntl);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+
+ cntl.Reset();
+ CallVersionByHttp(internal_ep, &cntl);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+
+ // The requests above did not go through the master service and their
+ // messages must not have been recycled as if they did, otherwise the pool
+ // of the master service is left with objects of another type in it.
+ for (int i = 0; i < 10; ++i) {
+ TestBaiduMasterService(channel, brpc::COMPRESS_TYPE_NONE);
+ }
+
+ ASSERT_EQ(0, server.Stop(0));
+ ASSERT_EQ(0, server.Join());
+}
+
void TestGenericCall(brpc::Channel& channel, brpc::ContentType content_type,
brpc::CompressType compress_type,
brpc::ChecksumType checksum_type) {
diff --git a/test/brpc_socket_unittest.cpp b/test/brpc_socket_unittest.cpp
index 0d198106..6e616cbc 100644
--- a/test/brpc_socket_unittest.cpp
+++ b/test/brpc_socket_unittest.cpp
@@ -118,6 +118,17 @@ class CheckRecycle : public brpc::SocketUser {
}
};
+// Acceptor::Join() returns as soon as Acceptor::BeforeRecycle() cleared
+// _listened_fd, which Socket::BeforeRecycled() calls several statements before
+// it closes the fd, so the fd may still be open when Join() returns.
+void WaitForClosedFd(int fd) {
+ for (int i = 0; i < 100 && fcntl(fd, F_GETFD) != -1; ++i) {
+ bthread_usleep(10000);
+ }
+ ASSERT_EQ(-1, fcntl(fd, F_GETFD));
+ ASSERT_EQ(EBADF, errno);
+}
+
// Never hardcode a port in tests: it may be taken by another process or by
// another test in this binary, making the test fail for unrelated reasons.
// Listen on port 0 and ask the kernel which port it picked instead.
@@ -443,8 +454,7 @@ TEST_F(SocketTest, single_threaded_connect_and_write) {
messenger->StopAccept(0);
messenger->Join();
ASSERT_EQ(-1, messenger->listened_fd());
- ASSERT_EQ(-1, fcntl(listening_fd, F_GETFD));
- ASSERT_EQ(EBADF, errno);
+ ASSERT_NO_FATAL_FAILURE(WaitForClosedFd(listening_fd));
// The socket object is likely to be reused,
// and the local side should be initialized.
@@ -819,8 +829,7 @@ TEST_F(SocketTest, health_check) {
messenger->StopAccept(0);
messenger->Join();
ASSERT_EQ(-1, messenger->listened_fd());
- ASSERT_EQ(-1, fcntl(listening_fd, F_GETFD));
- ASSERT_EQ(EBADF, errno);
+ ASSERT_NO_FATAL_FAILURE(WaitForClosedFd(listening_fd));
ASSERT_EQ(0, brpc::Socket::SetFailed(id));
// StartHealthCheck is possibly still addressing the Socket.
@@ -1560,8 +1569,7 @@ TEST_F(SocketTest, keepalive_input_message) {
messenger->StopAccept(0);
messenger->Join();
ASSERT_EQ(-1, messenger->listened_fd());
- ASSERT_EQ(-1, fcntl(listening_fd, F_GETFD));
- ASSERT_EQ(EBADF, errno);
+ ASSERT_NO_FATAL_FAILURE(WaitForClosedFd(listening_fd));
}
TEST_F(SocketTest, socket_buffer_options_before_connect) {
@@ -1728,8 +1736,7 @@ TEST_F(SocketTest, tcp_user_timeout) {
messenger->StopAccept(0);
messenger->Join();
ASSERT_EQ(-1, messenger->listened_fd());
- ASSERT_EQ(-1, fcntl(listening_fd, F_GETFD));
- ASSERT_EQ(EBADF, errno);
+ ASSERT_NO_FATAL_FAILURE(WaitForClosedFd(listening_fd));
}
#endif
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
