Setup the message bridge key sharing services
This implements 2 new services that will be used to share the active
SCTP authentication key. For context, if SCTP authentication is wanted,
then we will need a way to securely distribute a shared key across every
node. We use gRPC to distribute the key.
* message_bridge_auth_server
This service should only run in one node. It generates a 128-bit
random-key during initialization. It sets up the gRPC service using
mutual-TLS authentication.
* message_bridge_auth_client
This service will run in every node. It listens for requests
in /aos aos.message_bridge.SctpConfigRequest and requests the
active key from the gRPC server which gets propagated into /aos
aos.message_bridge.SctpConfig. message_bridge reads this value and sets
the authentication key (previous change in relation).
These have some additional side-effects:
* This change also forces us to bring in an appropriate version of GRPC,
which itself forces updates to protobuf and other dependencies. We
take the opportunity to remove the protobuf subtree in favor of
a more sanitized import in the WORKSPACE.
* The various upgrades also upgraded our version of buildifier,
which now performs more aggressive linting.
* Our version of abseil was upgraded, which now forces you to only
have access to const references to things stored in absl::btree_set's
(see https://github.com/abseil/abseil-cpp/commit/a74b796ab3f114f6991479c9ad9e4c1a0dad3a4b).
Change-Id: I870b8f93451056e011cefa3cdf3c5dc01c19a6f9
Signed-off-by: James Kuszmaul <james.kuszmaul@bluerivertech.com>
Signed-off-by: Adam Snaider <adsnaider@gmail.com>
diff --git a/aos/config.bzl b/aos/config.bzl
index faaf45a..baeb0d3 100644
--- a/aos/config.bzl
+++ b/aos/config.bzl
@@ -82,7 +82,7 @@
"config_binary": attr.output(mandatory = True),
"_config_flattener": attr.label(
executable = True,
- cfg = "host",
+ cfg = "exec",
default = Label("//aos:config_flattener"),
),
"src": attr.label(
diff --git a/aos/configuration.cc b/aos/configuration.cc
index 78f9eb2..64804e5 100644
--- a/aos/configuration.cc
+++ b/aos/configuration.cc
@@ -139,6 +139,26 @@
namespace configuration {
namespace {
+template <typename T>
+struct FbsContainer {
+ FbsContainer(aos::FlatbufferDetachedBuffer<T> table) {
+ this->table =
+ std::make_unique<FlatbufferDetachedBuffer<T>>(std::move(table));
+ }
+ std::unique_ptr<FlatbufferDetachedBuffer<T>> table;
+ bool operator==(const FbsContainer<T> &other) const {
+ return *this->table == *other.table;
+ }
+ bool operator<(const FbsContainer<T> &other) const {
+ return *this->table < *other.table;
+ }
+};
+
+typedef FbsContainer<Channel> ChannelContainer;
+typedef FbsContainer<Connection> ConnectionContainer;
+typedef FbsContainer<Application> ApplicationContainer;
+typedef FbsContainer<Node> NodeContainer;
+
// Extracts the folder part of a path. Returns ./ if there is no path.
std::string_view ExtractFolder(const std::string_view filename) {
auto last_slash_pos = filename.find_last_of("/\\");
@@ -635,7 +655,7 @@
// Store all the channels in a sorted set. This lets us track channels we
// have seen before and merge the updates in.
- absl::btree_set<FlatbufferDetachedBuffer<Channel>> channels;
+ absl::btree_set<ChannelContainer> channels;
if (config.message().has_channels()) {
auto_merge_config.mutable_message()->clear_channels();
@@ -658,21 +678,21 @@
if (!result.second) {
// Already there, so merge the new table into the original.
// Schemas merge poorly, so pick the newest one.
- if (result.first->message().has_schema() && c->has_schema()) {
- result.first->mutable_message()->clear_schema();
+ if (result.first->table->message().has_schema() && c->has_schema()) {
+ result.first->table->mutable_message()->clear_schema();
}
auto merged =
- MergeFlatBuffers(*result.first, RecursiveCopyFlatBuffer(c));
+ MergeFlatBuffers(*result.first->table, RecursiveCopyFlatBuffer(c));
if (merged.message().has_destination_nodes()) {
- absl::btree_set<FlatbufferDetachedBuffer<Connection>> connections;
+ absl::btree_set<ConnectionContainer> connections;
for (const Connection *connection :
*merged.message().destination_nodes()) {
auto connection_result =
connections.insert(RecursiveCopyFlatBuffer(connection));
if (!connection_result.second) {
- *connection_result.first =
- MergeFlatBuffers(*connection_result.first,
+ *connection_result.first->table =
+ MergeFlatBuffers(*connection_result.first->table,
RecursiveCopyFlatBuffer(connection));
}
}
@@ -682,10 +702,9 @@
flatbuffers::FlatBufferBuilder fbb;
fbb.ForceDefaults(true);
std::vector<flatbuffers::Offset<Connection>> connection_offsets;
- for (const FlatbufferDetachedBuffer<Connection> &connection :
- connections) {
+ for (const ConnectionContainer &connection : connections) {
connection_offsets.push_back(
- RecursiveCopyFlatBuffer(&connection.message(), &fbb));
+ RecursiveCopyFlatBuffer(&connection.table->message(), &fbb));
}
flatbuffers::Offset<
flatbuffers::Vector<flatbuffers::Offset<Connection>>>
@@ -699,13 +718,13 @@
}
}
- *result.first = std::move(merged);
+ *result.first->table = std::move(merged);
}
}
}
// Now repeat this for the application list.
- absl::btree_set<FlatbufferDetachedBuffer<Application>> applications;
+ absl::btree_set<ApplicationContainer> applications;
if (config.message().has_applications()) {
auto_merge_config.mutable_message()->clear_applications();
for (const Application *a : *config.message().applications()) {
@@ -716,16 +735,16 @@
auto result = applications.insert(RecursiveCopyFlatBuffer(a));
if (!result.second) {
if (a->has_args()) {
- result.first->mutable_message()->clear_args();
+ result.first->table->mutable_message()->clear_args();
}
- *result.first =
- MergeFlatBuffers(*result.first, RecursiveCopyFlatBuffer(a));
+ *result.first->table =
+ MergeFlatBuffers(*result.first->table, RecursiveCopyFlatBuffer(a));
}
}
}
// Now repeat this for the node list.
- absl::btree_set<FlatbufferDetachedBuffer<Node>> nodes;
+ absl::btree_set<NodeContainer> nodes;
if (config.message().has_nodes()) {
auto_merge_config.mutable_message()->clear_nodes();
for (const Node *n : *config.message().nodes()) {
@@ -735,8 +754,8 @@
auto result = nodes.insert(RecursiveCopyFlatBuffer(n));
if (!result.second) {
- *result.first =
- MergeFlatBuffers(*result.first, RecursiveCopyFlatBuffer(n));
+ *result.first->table =
+ MergeFlatBuffers(*result.first->table, RecursiveCopyFlatBuffer(n));
}
}
}
@@ -750,9 +769,9 @@
channels_offset;
{
::std::vector<flatbuffers::Offset<Channel>> channel_offsets;
- for (const FlatbufferDetachedBuffer<Channel> &c : channels) {
+ for (const ChannelContainer &c : channels) {
channel_offsets.emplace_back(
- RecursiveCopyFlatBuffer<Channel>(&c.message(), &fbb));
+ RecursiveCopyFlatBuffer<Channel>(&c.table->message(), &fbb));
}
channels_offset = fbb.CreateVector(channel_offsets);
}
@@ -762,9 +781,9 @@
applications_offset;
{
::std::vector<flatbuffers::Offset<Application>> applications_offsets;
- for (const FlatbufferDetachedBuffer<Application> &a : applications) {
+ for (const ApplicationContainer &a : applications) {
applications_offsets.emplace_back(
- RecursiveCopyFlatBuffer<Application>(&a.message(), &fbb));
+ RecursiveCopyFlatBuffer<Application>(&a.table->message(), &fbb));
}
applications_offset = fbb.CreateVector(applications_offsets);
}
@@ -774,9 +793,9 @@
nodes_offset;
{
::std::vector<flatbuffers::Offset<Node>> node_offsets;
- for (const FlatbufferDetachedBuffer<Node> &n : nodes) {
+ for (const NodeContainer &n : nodes) {
node_offsets.emplace_back(
- RecursiveCopyFlatBuffer<Node>(&n.message(), &fbb));
+ RecursiveCopyFlatBuffer<Node>(&n.table->message(), &fbb));
}
nodes_offset = fbb.CreateVector(node_offsets);
}
diff --git a/aos/events/logging/logfile_utils.cc b/aos/events/logging/logfile_utils.cc
index 02cff6d..0166b72 100644
--- a/aos/events/logging/logfile_utils.cc
+++ b/aos/events/logging/logfile_utils.cc
@@ -1797,7 +1797,7 @@
source_node_index_(configuration::SourceNodeIndex(parts().config.get())) {
}
-Message *MessageSorter::Front() {
+const Message *MessageSorter::Front() {
// Queue up data until enough data has been queued that the front message is
// sorted enough to be safe to pop. This may do nothing, so we should make
// sure the nothing path is checked quickly.
@@ -1970,10 +1970,10 @@
return p;
}
-Message *PartsMerger::Front() {
+const Message *PartsMerger::Front() {
// Return the current Front if we have one, otherwise go compute one.
if (current_ != nullptr) {
- Message *result = current_->Front();
+ const Message *result = current_->Front();
CHECK_GE(result->timestamp.time, last_message_time_);
VLOG(1) << this << " PartsMerger::Front for node " << node_name() << " "
<< *result;
@@ -1982,10 +1982,10 @@
// Otherwise, do a simple search for the oldest message, deduplicating any
// duplicates.
- Message *oldest = nullptr;
+ const Message *oldest = nullptr;
sorted_until_ = monotonic_clock::max_time;
for (MessageSorter &message_sorter : message_sorters_) {
- Message *msg = message_sorter.Front();
+ const Message *msg = message_sorter.Front();
if (!msg) {
sorted_until_ = std::min(sorted_until_, message_sorter.sorted_until());
continue;
@@ -2072,9 +2072,9 @@
return configuration::NodeName(configuration().get(), node());
}
-Message *BootMerger::Front() {
+const Message *BootMerger::Front() {
if (parts_mergers_[index_].get() != nullptr) {
- Message *result = parts_mergers_[index_]->Front();
+ const Message *result = parts_mergers_[index_]->Front();
if (result != nullptr) {
VLOG(1) << this << " BootMerger::Front " << node_name() << " " << *result;
@@ -2088,7 +2088,7 @@
return nullptr;
} else {
++index_;
- Message *result = Front();
+ const Message *result = Front();
VLOG(1) << this << " BootMerger::Front " << node_name() << " " << *result;
return result;
}
@@ -2199,7 +2199,7 @@
while (true) {
// Load all the timestamps. If we find data, ignore it and drop it on the
// floor. It will be read when boot_merger_ is used.
- Message *msg = timestamp_boot_merger_->Front();
+ const Message *msg = timestamp_boot_merger_->Front();
if (!msg) {
queue_timestamps_ran_ = true;
return;
@@ -2258,8 +2258,8 @@
timestamp_boot_merger_->monotonic_oldest_time(boot));
}
-Message *SplitTimestampBootMerger::Front() {
- Message *boot_merger_front = boot_merger_.Front();
+const Message *SplitTimestampBootMerger::Front() {
+ const Message *boot_merger_front = boot_merger_.Front();
if (timestamp_boot_merger_) {
CHECK(queue_timestamps_ran_);
@@ -2403,7 +2403,7 @@
}
}
-void TimestampMapper::QueueMessage(Message *msg) {
+void TimestampMapper::QueueMessage(const Message *msg) {
matched_messages_.emplace_back(
TimestampedMessage{.channel_index = msg->channel_index,
.queue_index = msg->queue_index,
@@ -2469,7 +2469,7 @@
if (nodes_data_.empty()) {
// Simple path. We are single node, so there are no timestamps to match!
CHECK_EQ(messages_.size(), 0u);
- Message *msg = boot_merger_.Front();
+ const Message *msg = boot_merger_.Front();
if (!msg) {
return MatchResult::kEndOfFile;
}
@@ -2745,7 +2745,7 @@
}
bool TimestampMapper::Queue() {
- Message *msg = boot_merger_.Front();
+ const Message *msg = boot_merger_.Front();
if (msg == nullptr) {
return false;
}
diff --git a/aos/events/logging/logfile_utils.h b/aos/events/logging/logfile_utils.h
index fb3fcbf..8a569ba 100644
--- a/aos/events/logging/logfile_utils.h
+++ b/aos/events/logging/logfile_utils.h
@@ -579,9 +579,8 @@
// The time this data is sorted until.
monotonic_clock::time_point sorted_until() const { return sorted_until_; }
- // Returns the next sorted message from the log file. It is safe to call
- // std::move() on the result to move the data flatbuffer from it.
- Message *Front();
+ // Returns the next sorted message from the log file.
+ const Message *Front();
// Pops the front message. This should only be called after a call to
// Front().
void PopFront();
@@ -656,9 +655,8 @@
// The time this data is sorted until.
monotonic_clock::time_point sorted_until() const { return sorted_until_; }
- // Returns the next sorted message from the set of log files. It is safe to
- // call std::move() on the result to move the data flatbuffer from it.
- Message *Front();
+ // Returns the next sorted message from the set of log files.
+ const Message *Front();
// Pops the front message. This should only be called after a call to
// Front().
void PopFront();
@@ -717,9 +715,8 @@
bool started() const;
- // Returns the next sorted message from the set of log files. It is safe to
- // call std::move() on the result to move the data flatbuffer from it.
- Message *Front();
+ // Returns the next sorted message from the set of log files.
+ const Message *Front();
// Pops the front message. This should only be called after a call to
// Front().
void PopFront();
@@ -787,9 +784,8 @@
return boot_merger_.started();
}
- // Returns the next sorted message from the set of log files. It is safe to
- // call std::move() on the result to move the data flatbuffer from it.
- Message *Front();
+ // Returns the next sorted message from the set of log files.
+ const Message *Front();
// Pops the front message. This should only be called after a call to
// Front().
@@ -975,7 +971,7 @@
void QueueUnmatchedUntil(BootTimestamp t);
// Queues m into matched_messages_.
- void QueueMessage(Message *m);
+ void QueueMessage(const Message *m);
// If a replay_channels_callback was set and the callback returns false, a
// matched message is popped and true is returned. Otherwise false is
diff --git a/aos/logging/BUILD b/aos/logging/BUILD
index d9e4e94..e179d6b 100644
--- a/aos/logging/BUILD
+++ b/aos/logging/BUILD
@@ -31,7 +31,6 @@
name = "log_namer",
srcs = ["log_namer.cc"],
hdrs = ["log_namer.h"],
- copts = ["-Wno-format-nonliteral"],
target_compatible_with = ["@platforms//os:linux"],
visibility = ["//visibility:public"],
deps = [
diff --git a/aos/logging/log_namer.cc b/aos/logging/log_namer.cc
index 3deb89d..4e9e00e 100644
--- a/aos/logging/log_namer.cc
+++ b/aos/logging/log_namer.cc
@@ -18,6 +18,12 @@
#include "aos/configuration.h"
+#if defined(__clang)
+#pragma clang diagnostic ignored "-Wformat-nonliteral"
+#elif defined(__GNUC__)
+#pragma GCC diagnostic ignored "-Wformat-nonliteral"
+#endif
+
DEFINE_string(logging_folder,
#ifdef AOS_ARCHITECTURE_arm_frc
"",
diff --git a/aos/network/BUILD b/aos/network/BUILD
index e8dd75f..7a8ceb5 100644
--- a/aos/network/BUILD
+++ b/aos/network/BUILD
@@ -3,6 +3,9 @@
load("//aos:flatbuffers.bzl", "cc_static_flatbuffer")
load("@com_github_google_flatbuffers//:build_defs.bzl", "flatbuffer_cc_library")
load("@com_github_google_flatbuffers//:typescript.bzl", "flatbuffer_ts_library")
+load("@com_github_grpc_grpc//bazel:cc_grpc_library.bzl", "cc_grpc_library")
+load("@rules_cc//cc:defs.bzl", "cc_proto_library")
+load("@rules_proto//proto:defs.bzl", "proto_library")
package(default_visibility = ["//visibility:public"])
@@ -137,6 +140,36 @@
visibility = ["//visibility:public"],
)
+proto_library(
+ name = "message_bridge_auth_proto",
+ srcs = [
+ "message_bridge_auth.proto",
+ ],
+ target_compatible_with = ["@platforms//os:linux"],
+ visibility = [":__pkg__"],
+)
+
+cc_proto_library(
+ name = "message_bridge_auth_proto_cc",
+ target_compatible_with = ["@platforms//os:linux"],
+ visibility = ["//visibility:public"],
+ deps = [
+ ":message_bridge_auth_proto",
+ ],
+)
+
+cc_grpc_library(
+ name = "message_bridge_auth_proto_grpc_cc",
+ srcs = [
+ ":message_bridge_auth_proto",
+ ],
+ grpc_only = True,
+ target_compatible_with = ["@platforms//os:linux"],
+ deps = [
+ ":message_bridge_auth_proto_cc",
+ ],
+)
+
cc_library(
name = "team_number",
srcs = [
@@ -173,10 +206,6 @@
hdrs = [
"sctp_lib.h",
],
- copts = [
- # The casts required to read datastructures from sockets trip -Wcast-align.
- "-Wno-cast-align",
- ],
target_compatible_with = ["@platforms//os:linux"],
deps = [
"//aos:unique_malloc_ptr",
@@ -214,9 +243,6 @@
hdrs = [
"sctp_server.h",
],
- copts = [
- "-Wno-cast-align",
- ],
target_compatible_with = ["@platforms//os:linux"],
deps = [
":sctp_lib",
@@ -285,9 +311,6 @@
hdrs = [
"message_bridge_server_lib.h",
],
- copts = [
- "-Wno-cast-align",
- ],
target_compatible_with = ["@platforms//os:linux"],
deps = [
":connect_fbs",
@@ -334,9 +357,6 @@
hdrs = [
"sctp_client.h",
],
- copts = [
- "-Wno-cast-align",
- ],
target_compatible_with = ["@platforms//os:linux"],
deps = [
":sctp_lib",
@@ -370,9 +390,6 @@
hdrs = [
"message_bridge_client_lib.h",
],
- copts = [
- "-Wno-cast-align",
- ],
target_compatible_with = ["@platforms//os:linux"],
deps = [
":connect_fbs",
@@ -397,9 +414,6 @@
srcs = [
"message_bridge_client.cc",
],
- copts = [
- "-Wno-cast-align",
- ],
target_compatible_with = ["@platforms//os:linux"],
deps = [
":message_bridge_client_lib",
@@ -413,6 +427,68 @@
],
)
+cc_library(
+ name = "message_bridge_auth_server_lib",
+ srcs = [
+ "message_bridge_auth_server_lib.cc",
+ ],
+ hdrs = [
+ "message_bridge_auth_server_lib.h",
+ ],
+ deps = [
+ ":message_bridge_auth_proto_cc",
+ ":message_bridge_auth_proto_grpc_cc",
+ "//aos/logging",
+ "//aos/util:file",
+ "@com_github_grpc_grpc//:grpc++",
+ "@com_google_absl//absl/strings",
+ ],
+)
+
+cc_library(
+ name = "message_bridge_auth_client_lib",
+ srcs = [
+ "message_bridge_auth_client_lib.cc",
+ ],
+ hdrs = [
+ "message_bridge_auth_client_lib.h",
+ ],
+ deps = [
+ ":message_bridge_auth_proto_grpc_cc",
+ ":message_bridge_client_fbs",
+ ":message_bridge_server_fbs",
+ ":sctp_config_fbs",
+ ":sctp_config_request_fbs",
+ "//aos/events:event_loop",
+ "//aos/logging",
+ "//aos/util:file",
+ "@com_github_grpc_grpc//:grpc++",
+ "@com_google_absl//absl/strings",
+ ],
+)
+
+cc_test(
+ name = "message_bridge_auth_test",
+ srcs = [
+ "message_bridge_auth_test.cc",
+ ],
+ data = [
+ ":message_bridge_auth_test_config",
+ ],
+ deps = [
+ ":message_bridge_auth_client_lib",
+ ":message_bridge_auth_server_lib",
+ "//aos:configuration",
+ "//aos/events:simulated_event_loop",
+ "//aos/logging",
+ "//aos/testing:googletest",
+ "//aos/testing:path",
+ "//aos/util:file",
+ "@com_github_grpc_grpc//:grpc++",
+ "@com_google_absl//absl/strings",
+ ],
+)
+
aos_config(
name = "message_bridge_test_combined_timestamps_common_config",
src = "message_bridge_test_combined_timestamps_common.json",
@@ -431,6 +507,17 @@
)
aos_config(
+ name = "message_bridge_auth_test_config",
+ src = "message_bridge_auth_test.json",
+ flatbuffers = [
+ ":sctp_config_fbs",
+ ":sctp_config_request_fbs",
+ ],
+ target_compatible_with = ["@platforms//os:linux"],
+ deps = ["//aos/events:aos_config"],
+)
+
+aos_config(
name = "message_bridge_test_common_config",
src = "message_bridge_test_common.json",
flatbuffers = [
@@ -772,7 +859,6 @@
srcs = [
"sctp_perf.cc",
],
- copts = ["-Wno-cast-align"],
deps = [
":sctp_client",
":sctp_server",
diff --git a/aos/network/message_bridge_auth.proto b/aos/network/message_bridge_auth.proto
new file mode 100644
index 0000000..178c51d
--- /dev/null
+++ b/aos/network/message_bridge_auth.proto
@@ -0,0 +1,16 @@
+syntax = "proto3";
+
+package aos.message_bridge;
+
+message SctpKeyResponse {
+ // Active SCTP authentication key.
+ bytes key = 1;
+}
+
+message SctpKeyRequest {}
+
+service SctpConfigService {
+ // Returns the active SCTP authentication key that should be used
+ // across nodes in message bridge.
+ rpc GetActiveKey(SctpKeyRequest) returns (SctpKeyResponse) {}
+}
diff --git a/aos/network/message_bridge_auth_client_lib.cc b/aos/network/message_bridge_auth_client_lib.cc
new file mode 100644
index 0000000..3b71632
--- /dev/null
+++ b/aos/network/message_bridge_auth_client_lib.cc
@@ -0,0 +1,62 @@
+#include "aos/network/message_bridge_auth_client_lib.h"
+
+#include <vector>
+
+#include "glog/logging.h"
+
+#include "aos/events/event_loop.h"
+#include "aos/network/message_bridge_auth.grpc.pb.h"
+#include "aos/network/message_bridge_client_generated.h"
+#include "aos/network/message_bridge_server_generated.h"
+#include "aos/network/sctp_config_generated.h"
+#include "aos/network/sctp_config_request_generated.h"
+#include "aos/util/file.h"
+#include "grpc/grpc.h"
+#include "grpcpp/channel.h"
+
+namespace aos::message_bridge::auth {
+
+namespace {
+using ::grpc::Channel;
+using ::grpc::ClientContext;
+using ::grpc::Status;
+} // namespace
+
+MessageBridgeAuthClient::MessageBridgeAuthClient(
+ EventLoop *const event_loop, std::shared_ptr<Channel> channel)
+ : sender_(event_loop->MakeSender<SctpConfig>("/aos")),
+ client_(SctpConfigService::NewStub(channel)) {
+ event_loop->MakeWatcher("/aos", [this](const SctpConfigRequest &stats) {
+ if (stats.request_key()) {
+ VLOG(1) << "Got SCTP authentication request from /aos";
+ SendKey();
+ }
+ });
+}
+
+void MessageBridgeAuthClient::SendKey() {
+ std::vector<uint8_t> key(GetSctpKey());
+ if (key.empty()) {
+ return;
+ }
+ Sender<SctpConfig>::Builder sender = sender_.MakeBuilder();
+ auto fb_key = sender.fbb()->CreateVector(std::move(key));
+ auto builder = sender.MakeBuilder<SctpConfig>();
+ builder.add_key(fb_key);
+ sender.CheckOk(sender.Send(builder.Finish()));
+}
+
+std::vector<uint8_t> MessageBridgeAuthClient::GetSctpKey() {
+ ClientContext context;
+ SctpKeyRequest request;
+ SctpKeyResponse response;
+ Status status = client_->GetActiveKey(&context, request, &response);
+ if (!status.ok()) {
+ LOG_EVERY_N(ERROR, 50)
+ << "Unable to retreive active SCTP authentication key from server";
+ return {};
+ }
+
+ return std::vector<uint8_t>(response.key().begin(), response.key().end());
+}
+} // namespace aos::message_bridge::auth
diff --git a/aos/network/message_bridge_auth_client_lib.h b/aos/network/message_bridge_auth_client_lib.h
new file mode 100644
index 0000000..2ff6d1c
--- /dev/null
+++ b/aos/network/message_bridge_auth_client_lib.h
@@ -0,0 +1,37 @@
+#ifndef AOS_NETWORK_MESSAGE_BRIGE_AUTH_CLIENT_LIB_H_
+#define AOS_NETWORK_MESSAGE_BRIGE_AUTH_CLIENT_LIB_H_
+
+#include "aos/events/event_loop.h"
+#include "aos/network/message_bridge_auth.grpc.pb.h"
+#include "aos/network/sctp_config_generated.h"
+#include "grpcpp/channel.h"
+
+namespace aos::message_bridge::auth {
+
+// See the explanation of underlying the architecture in
+// aos/network/message_bridge_auth_server_lib.h.
+
+// This class propagates the SCTP authentication key from the gRPC server to
+// message bridge. It listens to requests in the /aos SctpConfigRequest channel
+// and sends responses to /aos SctpConfig channel.
+class MessageBridgeAuthClient {
+ public:
+ MessageBridgeAuthClient(EventLoop *const event_loop,
+ std::shared_ptr<grpc::Channel> channel);
+
+ // Gets the key from the gRPC channel and sends it to the /aos SctpConfig
+ // channel.
+ void SendKey();
+
+ private:
+ // Gets the active SCTP key from the authentication server.
+ //
+ // Returns an empty vector on failure.
+ std::vector<uint8_t> GetSctpKey();
+
+ Sender<SctpConfig> sender_;
+ const std::unique_ptr<SctpConfigService::Stub> client_;
+};
+} // namespace aos::message_bridge::auth
+
+#endif // AOS_NETWORK_MESSAGE_BRIGE_AUTH_CLIENT_LIB_H_
diff --git a/aos/network/message_bridge_auth_server_lib.cc b/aos/network/message_bridge_auth_server_lib.cc
new file mode 100644
index 0000000..6a9dc7e
--- /dev/null
+++ b/aos/network/message_bridge_auth_server_lib.cc
@@ -0,0 +1,44 @@
+#include "aos/network/message_bridge_auth_server_lib.h"
+
+#include <cstdio>
+#include <fstream>
+
+#include "glog/logging.h"
+
+#include "grpc/grpc.h"
+#include "grpcpp/server_context.h"
+
+namespace aos::message_bridge::auth {
+
+namespace {
+
+using ::grpc::ServerContext;
+using ::grpc::Status;
+
+constexpr int kSctpKeySize = 16;
+
+std::string GenerateSecureRandomSequence(size_t count) {
+ std::ifstream rng("/dev/random", std::ios::in | std::ios::binary);
+ CHECK(rng) << "Unable to open /dev/random";
+ std::string out(count, 0);
+ rng.read(out.data(), count);
+ CHECK(rng) << "Couldn't read from random device";
+ rng.close();
+ return out;
+}
+} // namespace
+
+// This class implements the SctpConfigService. It securely generates an SCTP
+// authentication key and sends it to the clients that request it.
+SctpConfigServer::SctpConfigServer()
+ : active_key_(GenerateSecureRandomSequence(kSctpKeySize)) {}
+
+Status SctpConfigServer::GetActiveKey(ServerContext * /*context*/,
+ const SctpKeyRequest * /*request*/,
+ SctpKeyResponse *response) {
+ VLOG(1) << "Got request for SCTP authentication key";
+ response->set_key(active_key_);
+ return Status::OK;
+}
+
+} // namespace aos::message_bridge::auth
diff --git a/aos/network/message_bridge_auth_server_lib.h b/aos/network/message_bridge_auth_server_lib.h
new file mode 100644
index 0000000..acd5817
--- /dev/null
+++ b/aos/network/message_bridge_auth_server_lib.h
@@ -0,0 +1,48 @@
+#ifndef AOS_NETWORK_MESSAGE_BRIGE_AUTH_SERVER_LIB_H_
+#define AOS_NETWORK_MESSAGE_BRIGE_AUTH_SERVER_LIB_H_
+
+#include <string>
+
+#include "aos/network/message_bridge_auth.grpc.pb.h"
+#include "aos/network/message_bridge_auth.pb.h"
+#include "grpc/grpc.h"
+
+namespace aos::message_bridge::auth {
+
+// We use two services to share the active SCTP authentication key. For context,
+// if SCTP authentication is wanted, then we will need a way to securely
+// distribute a shared key across all the nodes. We use gRPC to distribute the
+// key.
+//
+// * message_bridge_auth_server
+//
+// This service should only run on a single node. It generates an n-bit
+// random-key during initialization. The server and client should use a mutal
+// authenticated channel. Without mutual authentication, anyone would be able to
+// acquire the authentication key which would defeat the purpose.
+//
+// * message_bridge_auth_client
+//
+// This service will run on all nodes. It listens for requests in /aos
+// aos.message_bridge.SctpConfigRequest and requests the active key from the
+// gRPC server which gets propagated into /aos aos.message_bridge.SctpConfig.
+// message_bridge reads this value and sets the authentication key (previous
+// change in relation).
+
+// This class implements the SctpConfigService. It securely generates an SCTP
+// authentication key and sends it to the clients that request it.
+class SctpConfigServer final : public SctpConfigService::Service {
+ public:
+ SctpConfigServer();
+
+ grpc::Status GetActiveKey(grpc::ServerContext *context,
+ const SctpKeyRequest *_request,
+ SctpKeyResponse *response) override;
+
+ private:
+ std::string active_key_;
+};
+
+} // namespace aos::message_bridge::auth
+
+#endif // AOS_NETWORK_MESSAGE_BRIGE_AUTH_SERVER_LIB_H_
diff --git a/aos/network/message_bridge_auth_test.cc b/aos/network/message_bridge_auth_test.cc
new file mode 100644
index 0000000..a360bdf
--- /dev/null
+++ b/aos/network/message_bridge_auth_test.cc
@@ -0,0 +1,123 @@
+#include <chrono>
+#include <memory>
+#include <thread>
+
+#include "gmock/gmock-matchers.h"
+#include "gtest/gtest.h"
+
+#include "aos/configuration.h"
+#include "aos/events/simulated_event_loop.h"
+#include "aos/network/message_bridge_auth_client_lib.h"
+#include "aos/network/message_bridge_auth_server_lib.h"
+#include "aos/network/sctp_config_generated.h"
+#include "aos/network/sctp_config_request_generated.h"
+#include "aos/testing/path.h"
+#include "grpc/grpc.h"
+#include "grpc/grpc_security.h"
+#include "grpcpp/channel.h"
+#include "grpcpp/create_channel.h"
+#include "grpcpp/server_builder.h"
+
+namespace aos::message_bridge::auth::testing {
+namespace {
+
+using ::aos::testing::ArtifactPath;
+using ::grpc::Channel;
+using ::grpc::Server;
+using ::grpc::ServerBuilder;
+using ::testing::ElementsAreArray;
+using ::testing::Not;
+
+using namespace ::std::chrono_literals;
+
+void RunServer() {
+ std::string server_address("127.0.0.1:1234");
+ SctpConfigServer service;
+ ServerBuilder builder;
+ builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
+ builder.RegisterService(&service);
+ std::unique_ptr<Server> auth_server(builder.BuildAndStart());
+ auth_server->Wait();
+}
+
+std::shared_ptr<Channel> MakeAuthClientChannel() {
+ std::string server_address = "127.0.0.1:1234";
+ std::shared_ptr<Channel> channel(
+ grpc::CreateChannel(server_address, grpc::InsecureChannelCredentials()));
+ // Give 10s deadline to connect to server or bail.
+ CHECK(channel->WaitForConnected(std::chrono::system_clock::now() + 10s))
+ << "Couldn't connect to auth server.";
+ return channel;
+}
+
+class MessageBridgeAuthTest : public ::testing::Test {
+ public:
+ MessageBridgeAuthTest()
+ : config_(aos::configuration::ReadConfig(
+ ArtifactPath("aos/network/message_bridge_auth_test_config.json"))),
+ event_loop_factory_(&config_.message()),
+ client_event_loop_(event_loop_factory_.MakeEventLoop("client")),
+ auth_client_(client_event_loop_.get(), MakeAuthClientChannel()),
+ mock_event_loop_(event_loop_factory_.MakeEventLoop("mock")),
+ request_key_sender_(
+ mock_event_loop_->MakeSender<SctpConfigRequest>("/aos")) {}
+
+ static void SetUpTestSuite() { std::thread(RunServer).detach(); }
+
+ private:
+ FlatbufferDetachedBuffer<Configuration> config_;
+ SimulatedEventLoopFactory event_loop_factory_;
+ std::unique_ptr<EventLoop> client_event_loop_;
+ MessageBridgeAuthClient auth_client_;
+
+ protected:
+ void RequestAuthKey() {
+ LOG(INFO) << "Requesting auth key";
+ auto sender = request_key_sender_.MakeBuilder();
+ auto builder = sender.MakeBuilder<SctpConfigRequest>();
+ builder.add_request_key(true);
+ sender.CheckOk(sender.Send(builder.Finish()));
+ }
+
+ void RunFor(distributed_clock::duration duration) {
+ event_loop_factory_.RunFor(duration);
+ }
+
+ std::unique_ptr<EventLoop> mock_event_loop_;
+ Sender<SctpConfigRequest> request_key_sender_;
+};
+
+// The "obvious" test for the message bridge authentication server/client.
+//
+// It spins up the server (done once for the entire test suite), creates a
+// client, and checks that we propagate the key from the server into the
+// appropriate AOS channel upon an a key request.
+TEST_F(MessageBridgeAuthTest, SmokeTest) {
+ // How many times we receive a new key.
+ int key_count = 0;
+ std::vector<uint8_t> auth_key;
+ mock_event_loop_->MakeWatcher("/aos", [&](const SctpConfig &config) {
+ if (config.has_key()) {
+ key_count++;
+ if (auth_key.empty()) {
+ auth_key.assign(config.key()->begin(), config.key()->end());
+ }
+ LOG(INFO) << "Got new auth key";
+ // Key shouldn't change as we are running on the same server.
+ EXPECT_THAT(auth_key,
+ ElementsAreArray(config.key()->begin(), config.key()->end()));
+ }
+ });
+
+ RunFor(1000ms);
+ RequestAuthKey();
+ RunFor(1000ms);
+ RequestAuthKey();
+ RunFor(1000ms);
+ EXPECT_FALSE(auth_key.empty());
+ // We requested the key twice, we should get it back twice.
+ EXPECT_EQ(key_count, 2);
+}
+
+} // namespace
+} // namespace aos::message_bridge::auth::testing
diff --git a/aos/network/message_bridge_auth_test.json b/aos/network/message_bridge_auth_test.json
new file mode 100644
index 0000000..8ebdda0
--- /dev/null
+++ b/aos/network/message_bridge_auth_test.json
@@ -0,0 +1,32 @@
+{
+ "channels": [
+ {
+ "name": "/aos",
+ "type": "aos.logging.LogMessageFbs",
+ "frequency": 200,
+ "num_senders": 20,
+ "max_size": 2048
+ },
+ {
+ "name": "/aos",
+ "type": "aos.timing.Report",
+ "frequency": 50,
+ "num_senders": 20,
+ "max_size": 2048
+ },
+ {
+ "name": "/aos",
+ "type": "aos.message_bridge.SctpConfig",
+ "frequency": 10,
+ "num_senders": 1,
+ "max_size": 256
+ },
+ {
+ "name": "/aos",
+ "type": "aos.message_bridge.SctpConfigRequest",
+ "frequency": 1,
+ "num_senders": 2,
+ "max_size": 32
+ }
+ ]
+}
diff --git a/aos/network/message_bridge_client_lib.cc b/aos/network/message_bridge_client_lib.cc
index 34f578e..bb233a9 100644
--- a/aos/network/message_bridge_client_lib.cc
+++ b/aos/network/message_bridge_client_lib.cc
@@ -19,6 +19,11 @@
#include "aos/unique_malloc_ptr.h"
#include "aos/util/file.h"
+// The casts required to read datastructures from sockets trip - Wcast - align.
+#ifdef __clang
+#pragma clang diagnostic ignored "-Wcast-align"
+#endif
+
DECLARE_bool(use_sctp_authentication);
// This application receives messages from another node and re-publishes them on
diff --git a/aos/network/message_bridge_server_lib.cc b/aos/network/message_bridge_server_lib.cc
index 35c8afe..4f582e7 100644
--- a/aos/network/message_bridge_server_lib.cc
+++ b/aos/network/message_bridge_server_lib.cc
@@ -17,6 +17,11 @@
#include "aos/network/sctp_server.h"
#include "aos/network/timestamp_channel.h"
+// The casts required to read datastructures from sockets trip - Wcast - align.
+#ifdef __clang
+#pragma clang diagnostic ignored "-Wcast-align"
+#endif
+
// For retrying sends on reliable channels, we will do an additive backoff
// strategy where we start at FLAGS_min_retry_period_ms and then add
// kRetryAdditivePeriod every time the retry fails, up until
diff --git a/aos/network/sctp_lib.cc b/aos/network/sctp_lib.cc
index 0922d6b..99a7f09 100644
--- a/aos/network/sctp_lib.cc
+++ b/aos/network/sctp_lib.cc
@@ -17,6 +17,11 @@
#include "aos/util/file.h"
+// The casts required to read datastructures from sockets trip - Wcast - align.
+#ifdef __clang
+#pragma clang diagnostic ignored "-Wcast-align"
+#endif
+
DEFINE_string(interface, "", "network interface");
DEFINE_bool(disable_ipv6, false, "disable ipv6");
DEFINE_int32(rmem, 0, "If nonzero, set rmem to this size.");
diff --git a/aos/network/sctp_perf.cc b/aos/network/sctp_perf.cc
index 5201f47..db762de 100644
--- a/aos/network/sctp_perf.cc
+++ b/aos/network/sctp_perf.cc
@@ -9,6 +9,11 @@
#include "aos/network/sctp_lib.h"
#include "aos/network/sctp_server.h"
+// The casts required to read datastructures from sockets trip - Wcast - align.
+#ifdef __clang
+#pragma clang diagnostic ignored "-Wcast-align"
+#endif
+
DEFINE_string(config, "aos_config.json", "Path to the config.");
DEFINE_uint32(port, 1323, "Port to run the sctp test on");
DEFINE_uint32(payload_size, 1000, "Size of data to send in bytes");
diff --git a/aos/seasocks/gen_embedded.bzl b/aos/seasocks/gen_embedded.bzl
index 6b01dac..3db1a7d 100644
--- a/aos/seasocks/gen_embedded.bzl
+++ b/aos/seasocks/gen_embedded.bzl
@@ -22,7 +22,7 @@
"_gen_embedded": attr.label(
executable = True,
default = Label("//aos/seasocks:gen_embedded"),
- cfg = "host",
+ cfg = "exec",
),
},
output_to_genfiles = True,
diff --git a/aos/util/BUILD b/aos/util/BUILD
index d12a10b..8450298 100644
--- a/aos/util/BUILD
+++ b/aos/util/BUILD
@@ -114,7 +114,7 @@
"//aos:flatbuffer_utils",
"//aos/events:event_loop",
"@com_github_lz4_lz4//:lz4",
- "@com_github_nlohmann_json//:json",
+ "@com_github_nlohmann_json//:nlohmann_json",
],
)
@@ -159,7 +159,7 @@
deps = [
":mcap_logger",
"//aos/testing:googletest",
- "@com_github_nlohmann_json//:json",
+ "@com_github_nlohmann_json//:nlohmann_json",
],
)
diff --git a/aos/util/mcap_logger.cc b/aos/util/mcap_logger.cc
index cdac928..df2b2d4 100644
--- a/aos/util/mcap_logger.cc
+++ b/aos/util/mcap_logger.cc
@@ -3,7 +3,7 @@
#include "absl/strings/str_replace.h"
#include "lz4/lz4.h"
#include "lz4/lz4frame.h"
-#include "single_include/nlohmann/json.hpp"
+#include "nlohmann/json.hpp"
#include "aos/configuration_schema.h"
#include "aos/flatbuffer_merge.h"
diff --git a/aos/util/mcap_logger.h b/aos/util/mcap_logger.h
index 8032c763..d9d6f36 100644
--- a/aos/util/mcap_logger.h
+++ b/aos/util/mcap_logger.h
@@ -1,7 +1,7 @@
#ifndef AOS_UTIL_MCAP_LOGGER_H_
#define AOS_UTIL_MCAP_LOGGER_H_
-#include "single_include/nlohmann/json.hpp"
+#include "nlohmann/json.hpp"
#include "aos/configuration_generated.h"
#include "aos/events/event_loop.h"
diff --git a/aos/vision/download/downloader.bzl b/aos/vision/download/downloader.bzl
index c972639..e76fe8e 100644
--- a/aos/vision/download/downloader.bzl
+++ b/aos/vision/download/downloader.bzl
@@ -41,7 +41,7 @@
attrs = {
"_downloader": attr.label(
executable = True,
- cfg = "host",
+ cfg = "exec",
default = Label("//aos/vision/download:downloader"),
),
"srcs": attr.label_list(
diff --git a/aos/vision/image/BUILD b/aos/vision/image/BUILD
index 63be82c..9e49326 100644
--- a/aos/vision/image/BUILD
+++ b/aos/vision/image/BUILD
@@ -1,4 +1,5 @@
-load("@com_google_protobuf//:protobuf.bzl", "cc_proto_library")
+load("@rules_cc//cc:defs.bzl", "cc_proto_library")
+load("@rules_proto//proto:defs.bzl", "proto_library")
package(default_visibility = ["//visibility:public"])
@@ -11,10 +12,15 @@
],
)
+proto_library(
+ name = "camera_params_proto",
+ srcs = ["camera_params.proto"],
+)
+
cc_proto_library(
name = "camera_params",
- srcs = ["camera_params.proto"],
target_compatible_with = ["@platforms//os:linux"],
+ deps = [":camera_params_proto"],
)
cc_library(