Writing gRPC Clients and Servers with C++20 Coroutines

The Callback Hell of C++ Async Programming

If you have ever written network programs in C++, you are probably very familiar with this scenario —

You need to send request B after request A completes, then send request C after B completes. So you end up writing code like this:

1
2
3
4
5
6
7
client.send(requestA, [](Response a) {
    client.send(requestB(a), [](Response b) {
        client.send(requestC(b), [](Response c) {
            // Finally made it here...
        });
    });
});

This is the infamous “callback hell”. The deeper the nesting, the harder the code is to read, and error handling becomes a nightmare — each layer of callback needs to handle errors independently. One missed callback call in a branch, and the entire request silently disappears without a trace.

The callback pattern also introduces another thorny problem: lifetime management. When a callback executes, every captured object must still be alive, yet their lifetimes are often hard to predict. You end up littering the code with shared_ptr and shared_from_this() just to keep things alive — ugly and unnecessarily expensive.

The most maddening part is cancellation. Imagine a user closes a page and you want to cancel an in-progress multi-step operation, but every step in the callback chain may have already started. How do you propagate the cancel signal? How do you ensure all resources are properly cleaned up? This is nearly an unsolvable problem.

If you have worked with async programming in Go or Python, you have probably envied their concise coroutine syntax — expressing asynchronous logic with synchronous-looking code. The good news is that C++20 introduced coroutines, and C++ programmers finally have a proper tool for the job.

The Async World of gRPC

Before diving in, let’s take a look at what gRPC’s async model looks like and what new challenges it brings.

The C++ implementation of gRPC provides two async APIs: the classic CompletionQueue-based model and the Reactor pattern callback model. Clients typically use the Reactor pattern — inheriting various ClientXxxReactor classes and implementing callbacks; servers typically use the CompletionQueue model — registering operations with the queue and polling for results.

For a client-side server-streaming RPC, you need to inherit grpc::ClientReadReactor<T> and implement several callbacks:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class MyReader : public grpc::ClientReadReactor<Response> {
public:
    void OnReadDone(bool ok) override {
        if (ok) {
            // Process the received data, then call StartRead again
            StartRead(&response_);
        }
    }

    void OnDone(const grpc::Status &status) override {
        // RPC finished
    }

private:
    Response response_;
};

The server side is even more complex. With the CompletionQueue-based model, you need to manually maintain a state machine:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
enum class State { WAIT, READ, WRITE, FINISH };

class Handler {
    void Proceed() {
        switch (state_) {
            case State::WAIT:
                // Register the next request, transition to READ state
                break;
            case State::READ:
                // Read data, decide whether to continue reading or writing
                break;
            // ...
        }
    }
};

This state machine must be maintained by hand. A single wrong state transition can cause data loss or a crash. Every time you need to implement this logic for a new RPC method, you have to go through the same ordeal.

Worse still, both approaches make it very hard to support cancellation cleanly. In the Reactor model, cancellation means calling context->TryCancel() and waiting for the OnDone callback; in the CompletionQueue model, state transitions after cancellation require extra care.

The root of the problem is: gRPC’s async API is designed around callbacks, while we want to organize code around the logical flow.

Promise and Future: Bridging Callbacks and Coroutines

This is exactly where asyncio shines.

asyncio is an async framework built on C++20 coroutines and the libuv event loop. Its core design idea is simple: use Promise and Future as a bridge to connect the callback world and the coroutine world.

A Promise is an object that can be “resolved” or “rejected”. Future is its other face — you can co_await a Future, and the coroutine will automatically resume when the Promise is resolved.

Take the sleep function as an example, to see how asyncio does it:

1
2
3
4
5
6
7
8
9
10
asyncio::task::Task<void> asyncio::sleep(std::chrono::milliseconds ms) {
    Promise<void> promise;

    uv_timer_start(timer, [](uv_timer_t *handle) {
        // Timer fires, resolve the promise, coroutine resumes
        static_cast<Promise<void> *>(handle->data)->resolve();
    }, ms.count(), 0);

    co_return co_await promise.getFuture();
}

Apply the same treatment to gRPC callbacks: resolve or reject a Promise inside the callback, and co_await the corresponding Future on the coroutine side. This lets you write what would otherwise require nested callbacks as straight-line code.

Cancellation Support

Coroutine cancellation is also implemented through the Promise mechanism. asyncio provides a Cancellable wrapper that takes a Future and a cancellation function:

1
2
3
4
5
6
7
8
co_return co_await asyncio::task::Cancellable{
    promise.getFuture(),
    [&]() -> std::expected<void, std::error_code> {
        // Perform the actual cancellation here
        context->TryCancel();
        return {};
    }
};

When external code calls task.cancel(), asyncio walks the task chain to find the Cancellable currently being awaited and executes its cancellation function. For gRPC, the cancellation function simply calls context->TryCancel(). gRPC then handles the cleanup, triggers the OnDone callback, the Promise is eventually rejected, and the coroutine ends with a cancellation error.

With this mechanism, we can cleanly add cancellation support to any gRPC async operation without introducing any special state variables into business code.

Defining the Sample Service

Before writing code, let’s define a sample service covering all four RPC types:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
syntax = "proto3";

package sample;

service SampleService {
  // Unary RPC: simplest request-response pattern
  rpc Echo(EchoRequest) returns (EchoResponse);

  // Server streaming: server continuously pushes data to the client
  rpc GetNumbers(GetNumbersRequest) returns (stream Number);

  // Client streaming: client continuously sends data, server returns one result
  rpc Sum(stream Number) returns (SumResponse);

  // Bidirectional streaming: both sides can send and receive continuously
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message EchoRequest  { string message = 1; }
message EchoResponse { string message = 1; int64 timestamp = 2; }
message GetNumbersRequest { int32 value = 1; int32 count = 2; }
message Number       { int32 value = 1; }
message SumResponse  { int32 total = 1; int32 count = 2; }
message ChatMessage  { string user = 1; string content = 2; int64 timestamp = 3; }

Each RPC pattern has its use: Echo is the canonical RPC, GetNumbers lets the server stream a batch of data, Sum lets the client stream data and get an aggregated result, and Chat is the most complex — a bidirectional stream where either side can send at any time.

Let’s implement them one by one, starting with the simplest: client Unary RPC.

Client Unary RPC

Unary RPC is the most straightforward: send one request, receive one response. The corresponding gRPC Reactor method signature is:

1
2
3
4
5
6
void Stub::async::Echo(
    grpc::ClientContext *,
    const EchoRequest *,
    EchoResponse *,
    std::function<void(grpc::Status)>
);

gRPC calls the callback we pass in when the RPC completes. Bridging it with Promise is natural:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
asyncio::task::Task<sample::EchoResponse> echo(
    const sample::EchoRequest &request,
    std::shared_ptr<grpc::ClientContext> context
) {
    sample::EchoResponse response;
    asyncio::Promise<void, std::string> promise;

    stub->async()->Echo(
        context.get(), &request, &response,
        [&](const grpc::Status &status) {
            if (!status.ok()) {
                promise.reject(status.error_message());
                return;
            }

            promise.resolve();
        }
    );

    if (const auto result = co_await asyncio::task::Cancellable{
        promise.getFuture(),
        [&]() -> std::expected<void, std::error_code> {
            context->TryCancel();
            return {};
        }
    }; !result)
        throw co_await asyncio::error::StacktraceError<std::runtime_error>::make(result.error());

    co_return response;
}

The core logic is just a few lines: construct a Promise, decide resolve or reject based on Status inside the callback, then co_await the Promise‘s Future.

Cancellation support is woven in naturally — wrap the Future with Cancellable, call context->TryCancel() on cancellation. From the caller’s perspective, this function is indistinguishable from a normal synchronous function, yet it never blocks the event loop.

Client Streaming RPCs

Streaming RPCs are more complex than Unary, because data is transmitted one piece at a time and each read or write is an independent async operation.

Server Streaming (Reader)

For a server-streaming RPC, the client needs to inherit grpc::ClientReadReactor<T>. Each call to StartRead causes gRPC to invoke OnReadDone when data is ready.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
template<typename T>
class Reader final : public grpc::ClientReadReactor<T> {
public:
    explicit Reader(std::shared_ptr<grpc::ClientContext> context) : mContext{std::move(context)} {
    }

    void OnDone(const grpc::Status &status) override {
        if (!status.ok()) {
            mDonePromise.reject(status.error_message());
            return;
        }

        mDonePromise.resolve();
    }

    void OnReadDone(const bool ok) override {
        // Resolve the per-read Promise when each read completes
        std::exchange(mReadPromise, std::nullopt)->resolve(ok);
    }

    asyncio::task::Task<std::optional<T>> read() {
        T element;

        asyncio::Promise<bool> promise;
        auto future = promise.getFuture();

        mReadPromise.emplace(std::move(promise));
        grpc::ClientReadReactor<T>::StartRead(&element);

        // ok == false means the stream has ended
        if (!co_await asyncio::task::Cancellable{
            std::move(future),
            [this]() -> std::expected<void, std::error_code> {
                mContext->TryCancel();
                return {};
            }
        })
            co_return std::nullopt;

        co_return element;
    }

    asyncio::task::Task<void> done() {
        if (const auto result = co_await mDonePromise.getFuture(); !result)
            throw co_await asyncio::error::StacktraceError<std::runtime_error>::make(result.error());
    }

private:
    std::shared_ptr<grpc::ClientContext> mContext;
    asyncio::Promise<void, std::string> mDonePromise;
    std::optional<asyncio::Promise<bool>> mReadPromise;
};

A few details worth noting:

  • mReadPromise is wrapped in std::optional because each call to read() constructs a new Promise.
  • std::exchange is used in OnReadDone to take ownership of the Promise, signaling that a single read has completed.
  • done() waits for the entire stream to finish; it holds a separate mDonePromise distinct from the per-read promise.

Using Reader to consume a server-streaming RPC:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
asyncio::task::Task<void> getNumbers(
    const sample::GetNumbersRequest &request,
    std::shared_ptr<grpc::ClientContext> context
) {
    Reader<sample::Number> reader{context};
    stub->async()->GetNumbers(context.get(), &request, &reader);

    reader.AddHold();
    reader.StartCall();

    while (true) {
        auto number = co_await reader.read();

        if (!number)
            break; // stream ended

        fmt::print("Received: {}\n", number->value());
    }

    reader.RemoveHold();
    co_await reader.done(); // wait for the stream to fully finish and check for errors
}

AddHold and RemoveHold are gRPC Reactor lifecycle control mechanisms that prevent the Reactor from being destroyed while we hold it.

Client Streaming (Writer)

Client-streaming RPC is similar to server-streaming but in the opposite direction. Inherit grpc::ClientWriteReactor<T>; after each StartWrite completes, OnWriteDone is called back:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
template<typename T>
class Writer final : public grpc::ClientWriteReactor<T> {
public:
    void OnWriteDone(const bool ok) override {
        std::exchange(mWritePromise, std::nullopt)->resolve(ok);
    }

    void OnWritesDoneDone(const bool ok) override {
        std::exchange(mWriteDonePromise, std::nullopt)->resolve(ok);
    }

    asyncio::task::Task<bool> write(const T element) {
        asyncio::Promise<bool> promise;
        auto future = promise.getFuture();

        mWritePromise.emplace(std::move(promise));
        grpc::ClientWriteReactor<T>::StartWrite(&element);

        co_return co_await asyncio::task::Cancellable{
            std::move(future),
            [this]() -> std::expected<void, std::error_code> {
                mContext->TryCancel();
                return {};
            }
        };
    }

    asyncio::task::Task<bool> writeDone() {
        asyncio::Promise<bool> promise;
        auto future = promise.getFuture();

        mWriteDonePromise.emplace(std::move(promise));
        grpc::ClientWriteReactor<T>::StartWritesDone();

        co_return co_await asyncio::task::Cancellable{
            std::move(future),
            [this]() -> std::expected<void, std::error_code> {
                mContext->TryCancel();
                return {};
            }
        };
    }

    // OnDone and done() are similar to Reader, omitted for brevity

private:
    std::shared_ptr<grpc::ClientContext> mContext;
    asyncio::Promise<void, std::string> mDonePromise;
    std::optional<asyncio::Promise<bool>> mWritePromise;
    std::optional<asyncio::Promise<bool>> mWriteDonePromise;
};

writeDone() corresponds to gRPC’s StartWritesDone, which signals to the server that the client has finished writing — equivalent to sending EOF on the stream.

Client Bidirectional Streaming

Bidirectional streaming is the most complex of the four patterns: the client simultaneously has both read and write capabilities. Fortunately, all that is needed is to merge the Reader and Writer logic into a single Stream class:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
template<typename RequestElement, typename ResponseElement>
class Stream final : public grpc::ClientBidiReactor<RequestElement, ResponseElement> {
public:
    // OnReadDone, OnWriteDone, OnWritesDoneDone, OnDone are the same as before

    asyncio::task::Task<std::optional<ResponseElement>> read() {
        ResponseElement element;

        asyncio::Promise<bool> promise;
        auto future = promise.getFuture();

        mReadPromise.emplace(std::move(promise));
        grpc::ClientBidiReactor<RequestElement, ResponseElement>::StartRead(&element);

        if (!co_await asyncio::task::Cancellable{
            std::move(future),
            [this]() -> std::expected<void, std::error_code> {
                mContext->TryCancel();
                return {};
            }
        })
            co_return std::nullopt;

        co_return element;
    }

    asyncio::task::Task<bool> write(const RequestElement element) { /* same as Writer */ }
    asyncio::task::Task<bool> writeDone()                         { /* same as Writer */ }
    asyncio::task::Task<void> done()                              { /* same as Reader */ }

private:
    std::shared_ptr<grpc::ClientContext> mContext;
    asyncio::Promise<void, std::string> mDonePromise;
    std::optional<asyncio::Promise<bool>> mReadPromise;
    std::optional<asyncio::Promise<bool>> mWritePromise;
    std::optional<asyncio::Promise<bool>> mWriteDonePromise;
};

When using a bidirectional stream, reading and writing are two independent tasks that run concurrently:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
asyncio::task::Task<void> chat(std::shared_ptr<grpc::ClientContext> context) {
    Stream<sample::ChatMessage, sample::ChatMessage> stream{context};
    stub->async()->Chat(context.get(), &stream);

    stream.AddHold();
    stream.StartCall();

    co_await all(
        // Read task
        asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
            while (const auto msg = co_await stream.read()) {
                fmt::print("Received: {}\n", msg->content());
            }
        }),
        // Write task
        asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
            sample::ChatMessage msg;
            msg.set_content("Hello!");
            co_await stream.write(msg);
            co_await stream.writeDone();
        })
    );

    stream.RemoveHold();
    co_await stream.done();
}

all() waits for both the read and write subtasks simultaneously. If either fails, it cancels the other and returns the error. This is the task-tree mechanism of asyncio in action — structured concurrency.

Wrapping GenericClient

The three streaming wrappers above all follow the same pattern, so it is time to unify them with templates. GenericClient provides one call overload for each of the four RPC types:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
template<typename T>
class GenericClient {
    using Stub = T::Stub;
    using AsyncStub = class Stub::async;

public:
    explicit GenericClient(std::unique_ptr<Stub> stub) : mStub{std::move(stub)} {
    }

protected:
    // 1. Unary RPC
    template<typename Request, typename Response>
    asyncio::task::Task<Response>
    call(
        void (AsyncStub::*method)(grpc::ClientContext *, const Request *, Response *,
                                   std::function<void(grpc::Status)>),
        std::shared_ptr<grpc::ClientContext> context,
        Request request
    ) {
        Response response;
        asyncio::Promise<void, std::string> promise;

        std::invoke(
            method,
            mStub->async(),
            context.get(),
            &request,
            &response,
            [&](const grpc::Status &status) {
                if (!status.ok()) {
                    promise.reject(status.error_message());
                    return;
                }

                promise.resolve();
            }
        );

        if (const auto result = co_await asyncio::task::Cancellable{
            promise.getFuture(),
            [&]() -> std::expected<void, std::error_code> {
                context->TryCancel();
                return {};
            }
        }; !result)
            throw co_await asyncio::error::StacktraceError<std::runtime_error>::make(result.error());

        co_return response;
    }

    // 2. Server streaming: push data into a channel via Sender
    template<typename Request, typename Element>
    asyncio::task::Task<void>
    call(
        void (AsyncStub::*method)(grpc::ClientContext *, const Request *,
                                   grpc::ClientReadReactor<Element> *),
        std::shared_ptr<grpc::ClientContext> context,
        Request request,
        asyncio::Sender<Element> sender
    ) {
        Reader<Element> reader{context};
        std::invoke(method, mStub->async(), context.get(), &request, &reader);

        reader.AddHold();
        reader.StartCall();

        const auto result = co_await asyncio::error::capture(
            asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
                while (true) {
                    auto element = co_await reader.read();

                    if (!element)
                        break;

                    co_await asyncio::error::guard(sender.send(*std::move(element)));
                }
            })
        );

        reader.RemoveHold();
        co_await reader.done();

        if (!result)
            std::rethrow_exception(result.error());
    }

    // 3. Client streaming: read data from Receiver and write into the stream
    template<typename Response, typename Element>
    asyncio::task::Task<Response>
    call(
        void (AsyncStub::*method)(grpc::ClientContext *, Response *,
                                   grpc::ClientWriteReactor<Element> *),
        std::shared_ptr<grpc::ClientContext> context,
        asyncio::Receiver<Element> receiver
    ) {
        Response response;
        Writer<Element> writer{context};
        std::invoke(method, mStub->async(), context.get(), &response, &writer);

        writer.AddHold();
        writer.StartCall();

        const auto result = co_await asyncio::error::capture(
            asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
                while (true) {
                    auto element = co_await receiver.receive();

                    if (!element) {
                        if (!co_await writer.writeDone())
                            fmt::print(stderr, "Write done failed\n");

                        if (element.error() != asyncio::ReceiveError::Disconnected)
                            throw co_await asyncio::error::StacktraceError<std::system_error>::make(element.error());

                        break;
                    }

                    co_await writer.write(*std::move(element));
                }
            })
        );

        writer.RemoveHold();
        co_await writer.done();

        if (!result)
            std::rethrow_exception(result.error());

        co_return response;
    }

    // 4. Bidirectional streaming: hold both a Receiver (input) and a Sender (output)
    template<typename RequestElement, typename ResponseElement>
    asyncio::task::Task<void>
    call(
        void (AsyncStub::*method)(grpc::ClientContext *,
                                   grpc::ClientBidiReactor<RequestElement, ResponseElement> *),
        std::shared_ptr<grpc::ClientContext> context,
        asyncio::Receiver<RequestElement> receiver,
        asyncio::Sender<ResponseElement> sender
    ) {
        Stream<RequestElement, ResponseElement> stream{context};
        std::invoke(method, mStub->async(), context.get(), &stream);

        stream.AddHold();
        stream.StartCall();

        const auto result = co_await asyncio::error::capture(
            all(
                asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
                    while (true) {
                        auto element = co_await stream.read();

                        if (!element)
                            break;

                        if (const auto res = co_await sender.send(*std::move(element)); !res) {
                            context->TryCancel();
                            throw co_await asyncio::error::StacktraceError<std::system_error>::make(res.error());
                        }
                    }
                }),
                asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
                    while (true) {
                        auto element = co_await receiver.receive();

                        if (!element) {
                            if (!co_await stream.writeDone())
                                fmt::print(stderr, "Write done failed\n");

                            if (element.error() != asyncio::ReceiveError::Disconnected)
                                throw co_await asyncio::error::StacktraceError<std::system_error>::make(element.error());

                            break;
                        }

                        co_await stream.write(*std::move(element));
                    }
                })
            )
        );

        stream.RemoveHold();
        co_await stream.done();

        if (!result)
            std::rethrow_exception(result.error());
    }

private:
    std::unique_ptr<Stub> mStub;
};

The four overloads are distinguished automatically by parameter types — the compiler selects the correct overload based on the method pointer type passed in. This is a classic use of template metaprogramming: different call patterns map to different function signatures, with zero runtime overhead.

For streaming RPCs, GenericClient uses asyncio::channel as the data conduit: Sender writes data into the channel, Receiver reads from it. The channel’s close signal (Receiver receiving a Disconnected error) naturally maps to stream EOF.

Implementing the Concrete Client

With GenericClient in place, implementing a concrete service client is straightforward:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class Client final : public GenericClient<sample::SampleService> {
public:
    using GenericClient::GenericClient;

    static Client make(const std::string &address) {
        return Client{sample::SampleService::NewStub(grpc::CreateChannel(address, grpc::InsecureChannelCredentials()))};
    }

    asyncio::task::Task<sample::EchoResponse>
    echo(
        sample::EchoRequest request,
        std::unique_ptr<grpc::ClientContext> context = std::make_unique<grpc::ClientContext>()
    ) {
        co_return co_await call(&sample::SampleService::Stub::async::Echo, std::move(context), std::move(request));
    }

    asyncio::task::Task<void>
    getNumbers(
        sample::GetNumbersRequest request,
        asyncio::Sender<sample::Number> sender,
        std::unique_ptr<grpc::ClientContext> context = std::make_unique<grpc::ClientContext>()
    ) {
        co_await call(
            &sample::SampleService::Stub::async::GetNumbers,
            std::move(context),
            std::move(request),
            std::move(sender)
        );
    }

    asyncio::task::Task<sample::SumResponse> sum(
        asyncio::Receiver<sample::Number> receiver,
        std::unique_ptr<grpc::ClientContext> context = std::make_unique<grpc::ClientContext>()
    ) {
        co_return co_await call(&sample::SampleService::Stub::async::Sum, std::move(context), std::move(receiver));
    }

    asyncio::task::Task<void>
    chat(
        asyncio::Receiver<sample::ChatMessage> receiver,
        asyncio::Sender<sample::ChatMessage> sender,
        std::unique_ptr<grpc::ClientContext> context = std::make_unique<grpc::ClientContext>()
    ) {
        co_return co_await call(
            &sample::SampleService::Stub::async::Chat,
            std::move(context),
            std::move(receiver),
            std::move(sender)
        );
    }
};

Here is how to call all four RPC types concurrently, showcasing the elegance of asyncio’s concurrent programming model:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
asyncio::task::Task<void> asyncMain(const int argc, char *argv[]) {
    auto client = Client::make("localhost:50051");

    co_await all(
        // Unary RPC
        asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
            sample::EchoRequest req;
            req.set_message("Hello gRPC!");
            const auto resp = co_await client.echo(req);
            fmt::print("Echo: {}\n", resp.message());
        }),

        // Server streaming + client streaming, connected via a channel
        asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
            sample::GetNumbersRequest req;
            req.set_value(1);
            req.set_count(5);

            auto [sender, receiver] = asyncio::channel<sample::Number>();

            const auto result = co_await all(
                client.getNumbers(req, std::move(sender)),
                client.sum(std::move(receiver))
            );

            const auto &resp = std::get<sample::SumResponse>(result);
            fmt::print("Sum: {}, count: {}\n", resp.total(), resp.count());
        }),

        // Bidirectional streaming
        asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
            auto [inSender, inReceiver] = asyncio::channel<sample::ChatMessage>();
            auto [outSender, outReceiver] = asyncio::channel<sample::ChatMessage>();

            co_await all(
                client.chat(std::move(outReceiver), std::move(inSender)),
                asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
                    sample::ChatMessage msg;
                    msg.set_content("Hello server!");
                    co_await asyncio::error::guard(outSender.send(std::move(msg)));
                    outSender.close();
                }),
                asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
                    const auto msg = co_await asyncio::error::guard(inReceiver.receive());
                    fmt::print("Chat reply: {}\n", msg.content());
                })
            );
        })
    );
}

The channel-based pipeline connecting getNumbers and sum is especially worth noting: numbers produced by the server-streaming RPC flow directly through the channel into the client-streaming RPC. The whole pipeline looks like synchronous code, but is fully asynchronous underneath.

Server Concurrency Model

The server has one extra dimension compared to the client: concurrency. A production-grade service must handle multiple client requests simultaneously, which requires the framework to dynamically manage an unknown number of concurrent tasks.

How CompletionQueue Works

The heart of the gRPC async server is CompletionQueue. The usage is roughly: when registering an async operation with gRPC, pass in a void *tag; when the operation completes, call CompletionQueue::Next() to retrieve the tag and learn whether the operation succeeded (bool ok):

1
2
3
4
5
6
7
void *tag{};
bool ok{};

while (cq->Next(&tag, &ok)) {
    // tag points to the object we passed in at registration; ok is the result
    dispatch(tag, ok);
}

But Next() is a blocking call — it blocks until an event arrives. Calling it directly on the event loop thread would freeze all of asyncio. The solution is to run it in a separate thread using asyncio::toThread:

1
2
3
4
5
6
7
8
9
asyncio::toThread([this] {
    void *tag{};
    bool ok{};

    while (mCompletionQueue->Next(&tag, &ok)) {
        // "Post" the completion event back to the event loop thread
        static_cast<asyncio::Promise<bool> *>(tag)->resolve(ok);
    }
})

Bridging with Promise

The other key question is: what does the tag point to? When registering an async operation, we pass the address of an asyncio::Promise<bool> as the tag:

1
2
3
4
5
6
7
8
9
10
11
asyncio::Promise<bool> promise;

service->RequestEcho(
    context.get(), &request, &writer,
    mCompletionQueue.get(), mCompletionQueue.get(),
    &promise   // <-- the tag is the address of the Promise
);

// Suspend the coroutine, wait for CompletionQueue notification
if (!co_await promise.getFuture())
    break; // ok == false means the CompletionQueue has shut down

When the operation completes, Next() returns tag = &promise, which then calls promise->resolve(ok). This resolve wakes up the coroutine suspended at co_await promise.getFuture(), seamlessly delivering the gRPC completion notification to the coroutine world.

The entire GenericServer::run() coordinates the two sides exactly this way:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
virtual asyncio::task::Task<void> run() {
    co_await all(
        dispatch(),                         // coroutine side: wait for and dispatch requests
        asyncio::toThread(
            [this] {                        // thread side: block-poll the CompletionQueue
                void *tag{};
                bool ok{};

                while (mCompletionQueue->Next(&tag, &ok)) {
                    static_cast<asyncio::Promise<bool> *>(tag)->resolve(ok);
                }
            }
        )
    );
}

dispatch() and toThread(...) run concurrently: the thread continuously pulls completion events from CompletionQueue and resolves the corresponding Promise; the coroutine side waits at co_await promise.getFuture() to be woken up, then handles a new request or a read/write completion each time. The two sides are decoupled through Promise and never block each other.

Dynamic Concurrency with TaskGroup

With the CompletionQueue bridge in place, the concurrency problem becomes clear. asyncio‘s TaskGroup is exactly what is needed to manage dynamically spawned concurrent tasks:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
asyncio::task::TaskGroup group;

while (true) {
    // Wait for the next request
    asyncio::Promise<bool> promise;
    service->RequestEcho(context, &request, &writer, cq, cq, &promise);

    if (!co_await promise.getFuture())
        break; // CompletionQueue has shut down

    // Spawn an independent handler task for this request
    auto task = asyncio::task::spawn(handleRequest(...));

    // Add to TaskGroup; the task removes itself automatically when it completes
    group.add(task);

    // Attach an error callback so unhandled exceptions are not silently lost
    task.future().fail([](const auto &e) {
        fmt::print(stderr, "Unhandled exception: {}\n", e);
    });
}

// On shutdown: cancel all in-flight requests and wait for them to finish
std::ignore = group.cancel();
co_await group;

A few important design decisions here:

  1. Accept loop decoupled from handler tasks: the loop waiting for new requests never blocks on handler logic, ensuring high concurrency.
  2. TaskGroup manages lifetimes: after each handler task joins the TaskGroup, a single cancel-and-wait call at shutdown drains all of them cleanly.
  3. Error isolation: a single request failure does not affect the rest of the service; errors are captured by the fail callback and logged, while other requests continue processing.

Server Unary RPC

With the concurrency model as background, implementing server Unary RPC is straightforward. The server uses grpc::ServerAsyncResponseWriter<Response> to send the response:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
asyncio::task::Task<void>
handleUnary(
    sample::SampleService::AsyncService *service,
    grpc::ServerCompletionQueue *cq
) {
    asyncio::task::TaskGroup group;

    while (true) {
        auto context = std::make_unique<grpc::ServerContext>();
        sample::EchoRequest request;
        grpc::ServerAsyncResponseWriter<sample::EchoResponse> writer{context.get()};

        {
            asyncio::Promise<bool> promise;
            service->RequestEcho(context.get(), &request, &writer, cq, cq, &promise);

            if (!co_await promise.getFuture())
                break;
        }

        auto task = asyncio::task::spawn(
            [
                context = std::move(context), request = std::move(request), writer = std::move(writer)
            ]() mutable -> asyncio::task::Task<void> {
                // Execute business logic
                const auto response = co_await asyncio::error::capture(echo(std::move(request)));

                asyncio::Promise<bool> promise;

                if (response) {
                    writer.Finish(*response, grpc::Status::OK, &promise);
                }
                else {
                    writer.FinishWithError(
                        grpc::Status{grpc::StatusCode::INTERNAL, fmt::to_string(response.error())},
                        &promise
                    );
                }

                // Wait for Finish to complete; cancellation is supported here too
                if (!co_await asyncio::task::Cancellable{
                    promise.getFuture(),
                    [&]() -> std::expected<void, std::error_code> {
                        context->TryCancel();
                        return {};
                    }
                })
                    throw co_await asyncio::error::StacktraceError<std::system_error>::make(
                        asyncio::task::Error::Cancelled
                    );
            }
        );

        group.add(task);
        task.future().fail([](const auto &e) {
            fmt::print(stderr, "Unhandled exception: {}\n", e);
        });
    }

    std::ignore = group.cancel();
    co_await group;
}

asyncio::error::capture is a utility that catches exceptions thrown by a coroutine and converts them to std::expected, enabling error handling with value semantics rather than exceptions. This way, even if business logic throws, we can gracefully convert it into a gRPC error status returned to the client instead of crashing the entire program.

Server Streaming RPCs

Server streaming RPCs require separate async wrappers for client-side reading and server-side writing. Unlike the client’s Reactor pattern, all server-side read/write operations go through CompletionQueue — pass a Promise address as the tag, and when the operation completes Next() retrieves it and calls resolve, delivering the completion notification to the coroutine.

Server-Side Reader (Client Streaming)

When the server reads data from the client, it uses grpc::ServerAsyncReader<Response, Element>. It has two type parameters: Response is the type returned to the client at the end, and Element is the type of each stream element read.

There is a subtle problem with this design: the Response type leaks into Reader itself, meaning different response types require different Reader instances. We work around this with type erasure — hide the concrete Response type inside Impl, and expose only the generic IReader interface to the outside:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
template<typename Element>
class Reader {
public:
    // IReader allows different underlying gRPC reader types to share one façade
    class IReader {
    public:
        virtual ~IReader() = default;
        virtual asyncio::task::Task<std::optional<Element>> read() = 0;
    };

    template<typename Response>
    class Impl : public IReader {
    public:
        Impl(
            std::shared_ptr<grpc::ServerContext> context,
            std::shared_ptr<grpc::ServerAsyncReader<Response, Element>> reader
        ) : mContext{std::move(context)}, mReader{std::move(reader)} {
        }

        asyncio::task::Task<std::optional<Element>> read() override {
            Element element;
            asyncio::Promise<bool> promise;

            mReader->Read(&element, &promise);

            if (!co_await asyncio::task::Cancellable{
                promise.getFuture(),
                [this]() -> std::expected<void, std::error_code> {
                    mContext->TryCancel();
                    return {};
                }
            }) {
                // Distinguish between "cancelled" and "stream ended"
                if (co_await asyncio::task::cancelled)
                    throw co_await asyncio::error::StacktraceError<std::system_error>::make(
                        asyncio::task::Error::Cancelled
                    );

                co_return std::nullopt;
            }

            co_return element;
        }

    private:
        std::shared_ptr<grpc::ServerContext> mContext;
        std::shared_ptr<grpc::ServerAsyncReader<Response, Element>> mReader;
    };

    explicit Reader(std::unique_ptr<IReader> reader) : mReader{std::move(reader)} {
    }

    asyncio::task::Task<std::optional<Element>> read() {
        co_return co_await mReader->read();
    }

private:
    std::unique_ptr<IReader> mReader;
};

Notice the co_await asyncio::task::cancelled check in the read() implementation — this is the key to distinguishing “reached end of stream (ok == false)” from “task was cancelled”. The former is a normal termination signal; the latter should propagate a cancellation error upward.

With Reader in place, the client-streaming request accept loop mirrors Unary: wait for a new request, construct a Reader and pass it to the business function, then send back the Response returned by the business function via reader->Finish():

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
asyncio::task::TaskGroup group;

while (true) {
    auto context = std::make_shared<grpc::ServerContext>();
    auto reader = std::make_shared<grpc::ServerAsyncReader<sample::SumResponse, sample::Number>>(context.get());

    {
        asyncio::Promise<bool> promise;
        service->RequestSum(context.get(), reader.get(), cq, cq, &promise);

        if (!co_await promise.getFuture())
            break;
    }

    auto task = asyncio::task::spawn(
        [context = std::move(context), reader = std::move(reader)]() -> asyncio::task::Task<void> {
            using Impl = Reader<sample::Number>::Impl<sample::SumResponse>;

            const auto response = co_await asyncio::error::capture(
                sum(Reader<sample::Number>{std::make_unique<Impl>(context, reader)})
            );

            asyncio::Promise<bool> promise;

            if (response) {
                reader->Finish(*response, grpc::Status::OK, &promise);
            }
            else {
                reader->FinishWithError(
                    grpc::Status{grpc::StatusCode::INTERNAL, fmt::to_string(response.error())},
                    &promise
                );
            }

            if (!co_await asyncio::task::Cancellable{
                promise.getFuture(),
                [&]() -> std::expected<void, std::error_code> {
                    context->TryCancel();
                    return {};
                }
            })
                throw co_await asyncio::error::StacktraceError<std::system_error>::make(
                    asyncio::task::Error::Cancelled
                );
        }
    );

    group.add(task);
    task.future().fail([](const auto &e) {
        fmt::print(stderr, "Unhandled exception: {}\n", e);
    });
}

std::ignore = group.cancel();
co_await group;

Server-Side Writer (Server Streaming)

For writing data from server to client, grpc::ServerAsyncWriter<T> is used:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
template<typename T>
class Writer {
public:
    Writer(std::shared_ptr<grpc::ServerContext> context, std::shared_ptr<grpc::ServerAsyncWriter<T>> writer)
        : mContext{std::move(context)}, mWriter{std::move(writer)} {
    }

    asyncio::task::Task<void> write(const T element) {
        asyncio::Promise<bool> promise;
        mWriter->Write(element, &promise);

        if (!co_await asyncio::task::Cancellable{
            promise.getFuture(),
            [this]() -> std::expected<void, std::error_code> {
                mContext->TryCancel();
                return {};
            }
        })
            throw co_await asyncio::error::StacktraceError<std::system_error>::make(
                asyncio::task::Error::Cancelled
            );
    }

private:
    std::shared_ptr<grpc::ServerContext> mContext;
    std::shared_ptr<grpc::ServerAsyncWriter<T>> mWriter;
};

The server-side Writer is slightly simpler than the client-side one — the server does not need to send WritesDone (that is the client’s privilege).

The original gRPC design has the stream end via the Writer’s Finish or FinishWithError, but we apply a similar redesign as for Reader.

The server-streaming request accept loop mirrors client streaming: wait for a new request, construct a Writer and pass it to the business function; after the business function returns, call writer->Finish() to end the stream:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
asyncio::task::TaskGroup group;

while (true) {
    auto context = std::make_shared<grpc::ServerContext>();

    sample::GetNumbersRequest request;
    auto writer = std::make_shared<grpc::ServerAsyncWriter<sample::Number>>(context.get());

    {
        asyncio::Promise<bool> promise;
        service->RequestGetNumbers(context.get(), &request, writer.get(), cq, cq, &promise);

        if (!co_await promise.getFuture())
            break;
    }

    auto task = asyncio::task::spawn(
        [
            context = std::move(context), request = std::move(request), writer = std::move(writer)
        ]() mutable -> asyncio::task::Task<void> {
            const auto result = co_await asyncio::error::capture(
                getNumbers(std::move(request), Writer{context, writer})
            );

            asyncio::Promise<bool> promise;

            if (result) {
                writer->Finish(grpc::Status::OK, &promise);
            }
            else {
                writer->Finish(
                    grpc::Status{grpc::StatusCode::INTERNAL, fmt::to_string(result.error())},
                    &promise
                );
            }

            if (!co_await asyncio::task::Cancellable{
                promise.getFuture(),
                [&]() -> std::expected<void, std::error_code> {
                    context->TryCancel();
                    return {};
                }
            })
                throw co_await asyncio::error::StacktraceError<std::system_error>::make(
                    asyncio::task::Error::Cancelled
                );
        }
    );

    group.add(task);
    task.future().fail([](const auto &e) {
        fmt::print(stderr, "Unhandled exception: {}\n", e);
    });
}

std::ignore = group.cancel();
co_await group;

Server Bidirectional Streaming

The server bidirectional stream wraps grpc::ServerAsyncReaderWriter<ResponseElement, RequestElement>, combining both read and write directions:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
template<typename RequestElement, typename ResponseElement>
class Stream {
public:
    Stream(
        std::shared_ptr<grpc::ServerContext> context,
        std::shared_ptr<grpc::ServerAsyncReaderWriter<ResponseElement, RequestElement>> stream
    ) : mContext{std::move(context)}, mStream{std::move(stream)} {
    }

    asyncio::task::Task<std::optional<RequestElement>> read() {
        RequestElement element;
        asyncio::Promise<bool> promise;

        mStream->Read(&element, &promise);

        if (!co_await asyncio::task::Cancellable{
            promise.getFuture(),
            [this]() -> std::expected<void, std::error_code> {
                mContext->TryCancel();
                return {};
            }
        }) {
            if (co_await asyncio::task::cancelled)
                throw co_await asyncio::error::StacktraceError<std::system_error>::make(
                    asyncio::task::Error::Cancelled
                );

            co_return std::nullopt;
        }

        co_return element;
    }

    asyncio::task::Task<void> write(const ResponseElement element) {
        asyncio::Promise<bool> promise;
        mStream->Write(element, &promise);

        if (!co_await asyncio::task::Cancellable{
            promise.getFuture(),
            [this]() -> std::expected<void, std::error_code> {
                mContext->TryCancel();
                return {};
            }
        })
            throw co_await asyncio::error::StacktraceError<std::system_error>::make(
                asyncio::task::Error::Cancelled
            );
    }

private:
    std::shared_ptr<grpc::ServerContext> mContext;
    std::shared_ptr<grpc::ServerAsyncReaderWriter<ResponseElement, RequestElement>> mStream;
};

The bidirectional streaming request accept loop follows the same familiar pattern: register with ServerAsyncReaderWriter, construct a Stream and pass it to the business function chat; after chat returns, call stream->Finish() to wrap up:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
asyncio::task::TaskGroup group;

while (true) {
    auto context = std::make_shared<grpc::ServerContext>();
    auto stream = std::make_shared<grpc::ServerAsyncReaderWriter<
        sample::ChatMessage, sample::ChatMessage>>(context.get());

    {
        asyncio::Promise<bool> promise;
        service->RequestChat(context.get(), stream.get(), cq, cq, &promise);

        if (!co_await promise.getFuture())
            break;
    }

    auto task = asyncio::task::spawn(
        [context = std::move(context), stream = std::move(stream)]() -> asyncio::task::Task<void> {
            const auto result = co_await asyncio::error::capture(
                chat(Stream{context, stream})
            );

            asyncio::Promise<bool> promise;

            if (result) {
                stream->Finish(grpc::Status::OK, &promise);
            }
            else {
                stream->Finish(
                    grpc::Status{grpc::StatusCode::INTERNAL, fmt::to_string(result.error())},
                    &promise
                );
            }

            if (!co_await asyncio::task::Cancellable{
                promise.getFuture(),
                [&]() -> std::expected<void, std::error_code> {
                    context->TryCancel();
                    return {};
                }
            })
                throw co_await asyncio::error::StacktraceError<std::system_error>::make(
                    asyncio::task::Error::Cancelled
                );
        }
    );

    group.add(task);
    task.future().fail([](const auto &e) {
        fmt::print(stderr, "Unhandled exception: {}\n", e);
    });
}

std::ignore = group.cancel();
co_await group;

Wrapping GenericServer

With all the components in place, GenericServer encapsulates the “accept request → dispatch handler → graceful shutdown” flow into four handle overloads, one per RPC type:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
template<typename T>
class GenericServer {
    using AsyncService = T::AsyncService;

protected:
    // Unary RPC: handler signature is Task<Response>(Request)
    template<typename Service, typename Request, typename Response, typename F>
        requires std::derived_from<AsyncService, Service> &&
        requires(F &&f, Request request) {
            { std::invoke(std::forward<F>(f), std::move(request)) } -> std::same_as<asyncio::task::Task<Response>>;
        }
    asyncio::task::Task<void>
    handle(
        void (Service::*method)(grpc::ServerContext *,
                                Request *,
                                grpc::ServerAsyncResponseWriter<Response> *,
                                grpc::CompletionQueue *,
                                grpc::ServerCompletionQueue *,
                                void *),
        F handler
    ) {
        asyncio::task::TaskGroup group;

        while (true) {
            auto context = std::make_unique<grpc::ServerContext>();
            Request request;
            grpc::ServerAsyncResponseWriter<Response> writer{context.get()};

            {
                asyncio::Promise<bool> promise;

                std::invoke(
                    method,
                    mService,
                    context.get(),
                    &request,
                    &writer,
                    mCompletionQueue.get(),
                    mCompletionQueue.get(),
                    &promise
                );

                if (!co_await promise.getFuture())
                    break;
            }

            auto task = asyncio::task::spawn(
                [
                    &handler, context = std::move(context), request = std::move(request), writer = std::move(writer)
                ]() mutable -> asyncio::task::Task<void> {
                    const auto response = co_await asyncio::error::capture(std::invoke(handler, std::move(request)));

                    asyncio::Promise<bool> promise;

                    if (response) {
                        writer.Finish(*response, grpc::Status::OK, &promise);
                    }
                    else {
                        writer.FinishWithError(
                            grpc::Status{grpc::StatusCode::INTERNAL, fmt::to_string(response.error())},
                            &promise
                        );
                    }

                    if (!co_await asyncio::task::Cancellable{
                        promise.getFuture(),
                        [&]() -> std::expected<void, std::error_code> {
                            context->TryCancel();
                            return {};
                        }
                    })
                        throw co_await asyncio::error::StacktraceError<std::system_error>::make(
                            asyncio::task::Error::Cancelled
                        );
                }
            );

            group.add(task);
            task.future().fail([](const auto &e) {
                fmt::print(stderr, "Unhandled exception: {}", e);
            });
        }

        std::ignore = group.cancel();
        co_await group;
    }

    // The server streaming, client streaming, and bidi streaming overloads
    // follow the exact same structure as Unary. The only differences are
    // the method pointer type and how the IO wrapper is constructed:
    //   server streaming → Writer{context, writer}                       + writer->Finish()
    //   client streaming → Reader<E>{make_unique<Impl>(context, reader)} + reader->Finish() / FinishWithError()
    //   bidi streaming   → Stream{context, stream}                       + stream->Finish()

    // Server streaming: handler signature is Task<void>(Request, Writer<Element>)
    template<typename Service, typename Request, typename Element, typename F>
        requires std::derived_from<AsyncService, Service> &&
        requires(F &&f, Request request, Writer<Element> writer) {
            {
                std::invoke(std::forward<F>(f), std::move(request), std::move(writer))
            } -> std::same_as<asyncio::task::Task<void>>;
        }
    asyncio::task::Task<void>
    handle(
        void (Service::*method)(grpc::ServerContext *,
                                Request *,
                                grpc::ServerAsyncWriter<Element> *,
                                grpc::CompletionQueue *,
                                grpc::ServerCompletionQueue *,
                                void *),
        F handler
    ) { /* same as Unary, omitted */ }

    // Client streaming: handler signature is Task<Response>(Reader<Element>)
    template<typename Service, typename Element, typename Response, typename F>
        requires std::derived_from<AsyncService, Service> &&
        requires(F &&f, Reader<Element> reader) {
            { std::invoke(std::forward<F>(f), std::move(reader)) } -> std::same_as<asyncio::task::Task<Response>>;
        }
    asyncio::task::Task<void>
    handle(
        void (Service::*method)(grpc::ServerContext *,
                                grpc::ServerAsyncReader<Response, Element> *,
                                grpc::CompletionQueue *,
                                grpc::ServerCompletionQueue *,
                                void *),
        F handler
    ) { /* same as Unary, omitted */ }

    // Bidirectional streaming: handler signature is Task<void>(Stream<Req, Resp>)
    template<typename Service, typename RequestElement, typename ResponseElement, typename F>
        requires std::derived_from<AsyncService, Service> &&
        requires(F &&f, Stream<RequestElement, ResponseElement> stream) {
            { std::invoke(std::forward<F>(f), std::move(stream)) } -> std::same_as<asyncio::task::Task<void>>;
        }
    asyncio::task::Task<void>
    handle(
        void (Service::*method)(grpc::ServerContext *,
                                grpc::ServerAsyncReaderWriter<ResponseElement, RequestElement> *,
                                grpc::CompletionQueue *,
                                grpc::ServerCompletionQueue *,
                                void *),
        F handler
    ) { /* same as Unary, omitted */ }

public:
    asyncio::task::Task<void> shutdown() {
        co_await asyncio::toThread([=, this] {
            mServer->Shutdown();
            mCompletionQueue->Shutdown();
        });
    }

    virtual asyncio::task::Task<void> run() {
        co_await all(
            dispatch(),
            asyncio::toThread(
                [this] {
                    void *tag{};
                    bool ok{};

                    while (mCompletionQueue->Next(&tag, &ok)) {
                        static_cast<asyncio::Promise<bool> *>(tag)->resolve(ok);
                    }
                }
            )
        );
    }

private:
    virtual asyncio::task::Task<void> dispatch() = 0;

    std::unique_ptr<grpc::Server> mServer;
    std::unique_ptr<AsyncService> mService;
    std::unique_ptr<grpc::ServerCompletionQueue> mCompletionQueue;
};

The requires constraints enforce correct handler signatures — the compiler checks at instantiation time, and if you pass a handler with the wrong signature you get a clear compile-time error instead of a runtime crash.

Implementing the Concrete Server

The final Server implementation simply inherits GenericServer and fills in the business logic and dispatch:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class Server final : public GenericServer<sample::SampleService> {
public:
    using GenericServer::GenericServer;

    static Server make(const std::string &address) {
        auto service = std::make_unique<sample::SampleService::AsyncService>();

        grpc::ServerBuilder builder;

        builder.AddListeningPort(address, grpc::InsecureServerCredentials());
        builder.RegisterService(service.get());

        auto completionQueue = builder.AddCompletionQueue();
        auto server = builder.BuildAndStart();

        return {std::move(server), std::move(service), std::move(completionQueue)};
    }

private:
    // Unary: return Response directly; errors are automatically converted to gRPC error status
    static asyncio::task::Task<sample::EchoResponse> echo(sample::EchoRequest request) {
        sample::EchoResponse response;
        response.set_message(request.message());
        response.set_timestamp(std::time(nullptr));
        co_return response;
    }

    // Server streaming: accept a Writer, write elements one by one
    static asyncio::task::Task<void>
    getNumbers(sample::GetNumbersRequest request, Writer<sample::Number> writer) {
        for (int i = 0; i < request.count(); ++i) {
            sample::Number number;
            number.set_value(request.value() + i);
            co_await writer.write(number);
        }
    }

    // Client streaming: accept a Reader, read and aggregate
    static asyncio::task::Task<sample::SumResponse> sum(Reader<sample::Number> reader) {
        int total{0}, count{0};
        while (const auto number = co_await reader.read()) {
            total += number->value();
            ++count;
        }
        sample::SumResponse response;
        response.set_total(total);
        response.set_count(count);
        co_return response;
    }

    // Bidirectional streaming: read one message, echo one back
    static asyncio::task::Task<void> chat(Stream<sample::ChatMessage, sample::ChatMessage> stream) {
        while (const auto message = co_await stream.read()) {
            sample::ChatMessage response;
            response.set_user("Server");
            response.set_timestamp(std::time(nullptr));
            response.set_content(fmt::format("Echo: {}", message->content()));
            co_await stream.write(response);
        }
    }

    // Bind method pointers to handlers and start the listening loop for each RPC
    asyncio::task::Task<void> dispatch() override {
        co_await all(
            handle(&sample::SampleService::AsyncService::RequestEcho, echo),
            handle(&sample::SampleService::AsyncService::RequestGetNumbers, getNumbers),
            handle(&sample::SampleService::AsyncService::RequestSum, sum),
            handle(&sample::SampleService::AsyncService::RequestChat, chat)
        );
    }
};

Business code and framework code are completely separated. The four static functions echo, getNumbers, sum, and chat only need to care about their own logic: accept request parameters, perform IO through Reader/Writer/Stream, and return results. Everything else — waiting for new requests, handling concurrency, writing back status at completion, graceful shutdown — is handled entirely by GenericServer.

Graceful Shutdown

In production, a process must ensure all in-progress RPCs are properly handled before exiting. The graceful shutdown pattern with asyncio + gRPC looks like this:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
asyncio::task::Task<void> asyncMain(const int argc, char *argv[]) {
    auto server = Server::make("0.0.0.0:50051");
    auto signal = asyncio::Signal::make();

    co_await race(
        asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
            asyncio::sync::Event event;

            co_await asyncio::task::Cancellable{
                all(
                    server.run(),
                    asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
                        co_await asyncio::error::guard(event.wait());
                        co_await server.shutdown(); // notify the gRPC server to shut down
                    })
                ),
                [&]() -> std::expected<void, std::error_code> {
                    event.set(); // trigger the shutdown sequence
                    return {};
                }
            };
        }),
        asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
            co_await asyncio::error::guard(signal.on(SIGINT));
        })
    );
}

The full shutdown sequence:

  1. SIGINT is received; the signal-watching task inside race completes.
  2. race cancels the other task, triggering the Cancellable cancellation function, which calls event.set().
  3. event.wait() returns, and server.shutdown() is called — it runs mServer->Shutdown() and mCompletionQueue->Shutdown() in a thread pool, telling gRPC to stop accepting new requests and close the CompletionQueue.
  4. Once CompletionQueue is closed, the thread-pool task in GenericServer::run() returns, and server.run() completes.
  5. Meanwhile, the accept loops in each handle() method exit because promise.getFuture() returns false, then cancel all in-flight handler tasks in the TaskGroup and wait for them to finish.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
asyncio::task::Task<void> shutdown() {
    co_await asyncio::toThread([=, this] {
        mServer->Shutdown();
        mCompletionQueue->Shutdown();
    });
}

virtual asyncio::task::Task<void> run() {
    co_await all(
        dispatch(),
        asyncio::toThread(
            [this] {
                void *tag{};
                bool ok{};
                // Block on Next(), forwarding each completion event to the corresponding Promise
                while (mCompletionQueue->Next(&tag, &ok))
                    static_cast<asyncio::Promise<bool> *>(tag)->resolve(ok);
            }
        )
    );
}

asyncio::toThread moves blocking operations to the thread pool, ensuring the event loop thread is never blocked. CompletionQueue::Next is blocking and must run in a separate thread; Server::Shutdown may also block and is likewise offloaded.

Summary

Looking back over the journey, it is clear what asyncio fundamentally changes about gRPC async programming:

Callbacks become coroutines. The Promise/Future pair is the cornerstone of the entire approach. Every gRPC async callback — OnReadDone, OnWriteDone, OnDone, and the tags returned from CompletionQueue — is uniformly converted into resolve/reject calls on a Promise. The coroutine side needs only co_await.

Cancellation support with zero business-code intrusion. Through the Cancellable wrapper, every async await point naturally gains cancellation capability. Business code does not need to carry a context parameter or check a cancellation flag. The cancellation signal propagates automatically along the task chain until it reaches the waiting gRPC operation and calls TryCancel().

TaskGroup solves dynamic concurrency. The server must handle an unknown number of concurrent requests simultaneously. TaskGroup allows tasks to be added dynamically, cancelled in bulk, and awaited in bulk — exactly the right solution for this scenario.

Template constraints improve safety. GenericClient and GenericServer use overloading and requires constraints so that the compiler verifies handler signatures at instantiation time, turning potential runtime errors into compile-time failures.

Graceful shutdown without boilerplate. The event + race + Cancellable combination cleanly implements the full “receive signal → trigger shutdown → wait for all tasks” sequence, with almost no extra state variables.

The final business code demonstrates the value of all this: echo, getNumbers, sum, and chat read like ordinary synchronous functions — no nested callbacks, no state machines, no explicit lifetime management. Yet behind them is a complete, production-ready async gRPC service with high concurrency, cancellation support, and graceful shutdown.

This is the promise of coroutines: make async code as readable as sync code, while retaining all the performance advantages of async execution.

Full source code: GitHub

我的家庭网络架构

前言

很久没写博客了，最近正好有空可以谈谈我最近更新的家庭网络架构。我所有的网络组件都部署在一台 J1900 的机器上，安装的操作系统是 debian 10，主板配有四个网口。我将其中三个网口合并为一个 bridge 用来构建我的私有子网，供主机和路由器插线连接，另一个网口连接光猫作为出口。

科学上网

通过在软路由上使用 iptables + tproxy 重定向 TCP/UDP 流量到 clash，可以实现全局翻墙，所有设备无需设置。在讨论我遇到的问题之前，有必要了解一下 clash 在软路由上怎么进行规则判断的，换言之 clash 怎么知道我们当前的连接是否需要走代理。

DNS

软路由上的流量被重定向后，clash 只能获取的目标 ip，但是根据 ip 地区判断是远远不够的，那么该如何获取 ip 对应的域名呢？clash 自带了一个 DNS server，在软路由上将所有 DNS 请求转发给它，那么它就能记录域名和 ip 的对应关系。但是由于存在许多问题，这种 redir-host 模式在较新的 clash 中被删除了。它最大的问题就是无法应对 DNS 污染，例如最上游 DNS 被污染时，返回 google 的 ip 是 127.0.0.1 这种，那么该请求直接在主机上请求失败了。

FakeIP

在使用 redir-host 的过程中，我经常遇到网络不正常的情况，迫于无奈只能切换到 fake-ip 模式。它的原理就是 clash 的 DNS server 返回一个虚假的内网 ip，例如 198.168.0.0/16，同时记录 ip 与 hostname 的对应关系，这样就可以避免 DNS 污染的问题。当然 fake-ip 也有缺陷，例如 QQ 的某些域名指向就是 127.0.0.1，因为需要和 QQ 客户端通信，另外 Windows 的网络探测服务也无法在 fake-ip 模式下工作。对于这些特殊的域名，我们可以在 clash 配置中过滤掉，让它们返回真实的查询结果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
dns:
  enable: true
  ipv6: false
  listen: 0.0.0.0:5300
  enhanced-mode: fake-ip
  fake-ip-range: 198.18.0.1/16
  nameserver:
    - https://doh.pub/dns-query
    - https://dns.alidns.com/dns-query
  fake-ip-filter:
    - '*.lan'
    - 'wspeed.qq.com'
    - 'internal-api-lark-file.feishu.cn'
    - 'jrlt.beacon.qq.com'
    - 'localhost.ptlogin2.qq.com'
    - '+.srv.nintendo.net'
    - '+.stun.playstation.net'
    - '+.msftconnecttest.com'
    - '+.msftncsi.com'
    - '+.xboxlive.com'
    - 'msftconnecttest.com'
    - 'xbox.*.microsoft.com'
    - '*.battlenet.com.cn'
    - '*.battlenet.com'
    - '*.blzstatic.cn'
    - '*.battle.net'

游戏加速

游戏加速说到底还是代理，只不过是延迟很低的线路，使用小米等路由器同时开启翻墙和游戏加速插件肯定会有冲突，但是在自己配置的 Linux 路由器里，我可以精准地控制每一条规则。

docker

一年前我的方案是将 openwrt 版本的 UU加速器 跑在 docker 里，它其实是一个为主机加速提供的方案，所以我写了个 python 脚本将我的软路由伪装成 PS4：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#!/usr/bin/env python3
import socket

UDP_IP = "0.0.0.0"
UDP_PORT = 987

sock = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
sock.bind((UDP_IP, UDP_PORT))

while True:
    data, addr = sock.recvfrom(1024)
    print("received message from %s[%d]: %s" % (addr[0], addr[1], data.decode('utf8')))

    response = "HTTP/1.1 200 OK\r\nhost-id:0123456789AB\r\nhost-type:PS4\r\nhost-name:MyPS4\r\nhost-request-port:%d\r\ndevice-discovery-protocol-version:00020020\r\nsystem-version:07020001\r\nrunning-app-name:Youtube\r\nrunning-app-titleid:CUSA01116\r\n\r\n" % addr[1]
    sock.sendto(response.encode('utf8'), addr)

想象一下，UU加速器 在容器 172.17.0.3 中广播 UDP 包，然后这个脚本在软路由的 172.17.0.1 上收到了这个包，并成功回复让对方认为自己是一台 PS4。然后在手机 APP 上就可以开启加速，这个时候我们只需要在软路由上将需要加速的包路由给 172.17.0.3 即可，流量会顺利地从容器的 tun 网卡进入 UU加速器 的专线。

1
2
3
# https://github.com/Hackerl/docker-uuplugin
docker build -t docker-uuplugin . --build-arg UU_IP=172.17.0.3
docker run -d -p 16363:16363 --cap-add NET_ADMIN --device /dev/net/tun:/dev/net/tun -it docker-uuplugin

启动后添加路由规则：

1
2
ip rule add fwmark 0x163 table 0x163
ip route add default via 172.17.0.3 table 0x163

这个路由规则的意思是如果流量带有 0x163 标志，那么就路由给 172.17.0.3，接着我们可以在 iptables 里给特定流量打上标志：

1
2
3
4
5
6
iptables -t mangle -N UUPLUGIN
iptables -t mangle -A UUPLUGIN -p udp -s 192.168.10.0/24 --dport 32768:65534 -j MARK --set-mark 0x163
iptables -t mangle -A UUPLUGIN -p udp -s 192.168.10.0/24 --dport 10000:32768 -j MARK --set-mark 0x163
iptables -t mangle -A UUPLUGIN -p udp -s 192.168.10.0/24 --dport 1025:9999 -j MARK --set-mark 0x163
iptables -t mangle -A UUPLUGIN -m mark --mark 0x163 -j ACCEPT
iptables -t mangle -A PREROUTING -j UUPLUGIN

例如我们给 UDP 高端口号的包打上标志，这其实是大多数网络游戏的通讯方式，例如英雄联盟。

虚拟机

使用 docker 运行 UU加速器 进行游戏加速也存在问题，一就是可选择的节点并不多，毕竟这是主机加速方案，二就是它的 iptables 限制了只代理某些源 ip。由于第二个限制，我们没有办法直接在它的容器中部署 socks5 代理，以集成到 clash 规则中，当然我可以修改它的规则，但是每次启动加速都要修改的话太麻烦了。此时我有了一个大胆的想法，我可不可以在软路由上跑一个 windows 虚拟机，然后将流量转发到虚拟机中去加速，一想到 J1900 廉价的性能我心里就充满了问号。
我首先安装了一个 windows 7 虚拟机，连上 VNC 后果然很卡，然后部署了一个简单的 socks5 代理，接着将英雄联盟台服的域名全部抠出来，写好 clash 规则：

1
2
3
4
5
6
7
8
9
10
11
12
13
script:
  engine: expr
  shortcuts:
    game: network == 'udp' and dst_port >= 1025

rules:
  - SCRIPT,game,GAME
  - DOMAIN,lolstatic-a.akamaihd.net,GAME
  - DOMAIN-KEYWORD,riot,GAME
  - DOMAIN-SUFFIX,pvp.net,GAME
  - DOMAIN-SUFFIX,leagueoflegends.com,GAME
  - DOMAIN-SUFFIX,newrelic.com,GAME
  - DOMAIN,lolesports.com,GAME

尝试了一下确实可以工作，所有英雄联盟相关的流量都由虚拟机代理。那么接下来要做的就是怎么把这台虚拟机压缩一下，J1900 的确有点带不动，而且为了一个游戏加速我也并不是很乐意耗费这么多 CPU 和内存。我在网上找到了一个 Windows 7 Super-Nano Lite，镜像居然只有 316MB，确定这可以跑起来吗？我安装了一下确实跑起来了，但是包括网络驱动之类的东西全部被删除了。

磁盘占用只有 600MB，内存也只用了 200MB，为了进一步优化性能，我使用 virt-install 安装虚拟机时设置的网络和磁盘驱动都是 virtio：

1
2
wget https://fedorapeople.org/groups/virt/virtio-win/direct-downloads/archive-virtio/virtio-win-0.1.173-4/virtio-win-0.1.173_x86.vfd
sudo virt-install --name=win7 --virt-type kvm --memory 1024 --vcpus=1 --os-type=windows --os-variant win7 --disk path=win7.qcow2,format=qcow2,bus=virtio --cdrom Windows7SuperNanoLite.iso --graphics vnc,listen=0.0.0.0 --network default,model=virtio --disk path=virtio-win-0.1.173_x86.vfd,device=floppy --noautoconsole

在使用 VNC 安装的时候会提示找不到驱动器，因为 windows 7 无法识别 virtio 设备，所有在界面上点击加载驱动，加载 virtio-win-0.1.173_x86.vfd 软盘中的磁盘和网络驱动。安装后的系统是无法显示中文的，因为中文包也被删除了。找一台完整的 windows 7 机器，拷贝 C:\Windows\System32\C_936.NLS 以及 C:\Windows\Fonts\msyh.ttf 到虚拟机中，中文就不会乱码了。安装完加速器后，需要部署一个支持 TCP/UDP 的 socks5 代理，我找了一圈发现没几个支持 UDP 的，最后选择了 3proxy，配置文件如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
#!/usr/local/bin/3proxy

system "echo 3proxy up!"

########### SERVICE ###########
# Set timeouts
timeouts 1 5 30 60 180 1800 15 60

# Service installation, daemon for nix, service for win32
#daemon
service

########### LOGGING ###########
# Set up logs
#log "/var/logs/3proxy/%Y%m%d.log" D
log "3proxy-%Y%m%d.log" D
logformat "- +_L%t.%. %Y-%m-%d  %N.%p %E %U %C:%c %R:%r %O %I %h %T"
archiver rar rar a -df -inul %A %F
rotate 30

########### IFACE ###########
# External is the interface you will send data out from, set with a static IP
external 0.0.0.0
# Internal is the interface you will listen on, in this case localhost (no physical nic)
internal 0.0.0.0

########### SOCKS ###########
# Socks5  proxy setup
auth none
flush
maxconn 300
socks

搞完这个之后在英雄联盟台服开了把大乱斗，延迟居然才 30 比国服还低，看了下软路由的资源占用，虚拟机低负载时 qemu 进程稳定占用 45% 左右的单核 CPU，算是可以接受的范围，毕竟它只是 J1900 啊！有了这个比较稳定的虚拟方案后，其实可以将一些 windows 独占的软件放到里面，通过软路由上的 smb 共享文件，但是我现在使用的百度云、迅雷都是使用的 docker + noVNC 方案，暂时没有其余想折腾的东西了。

Nokia 7P 刷机

前言

我有一台老旧的 Nokia 7P 手机，购买于 2018 年暑假，当时看中的主要是它的类原生系统，它也在后续的工作中陪伴了我 4 年。虽然现在已经换成了苹果，但是平时看视频之类的娱乐还是停留在它上面。之前就一直想在这台 Nokia 7P 上安装第三方系统，但是由于尝试解锁 bootloader 一直失败，只能暂且作罢。
最近在网上看到了一篇解锁 bootloader 的文章，貌似已经有三方的工具公开了，随即摩拳擦掌跃跃欲试，挑了个空闲的日子开始这趟刷机之旅。

驱动

在诺基亚关机状态下，长按电源键与音量加可以进入 bootloader 模式，此时可以使用 Android SDK 提供的 fastboot 工具刷入 recovery，比如常见的 TWRP，这是刷机的首要步骤。然而将 bootloader 模式下的手机使用 USB 连接电脑，电脑是无法识别的，因为缺少了驱动程序。
安装驱动程序最简单的方式就是在手机开机状态下，使用 USB 连接电脑，会自动从手机挂载出一个 CD-ROM 盘符，里面就有官方驱动安装程序，更多的安装方式可以参考这篇文章。
安装完驱动后打开设备管理器，如果能看到 bootloader 状态下的手机被识别成 fastboot 设备，则说明驱动已正常运行。

解锁

在驱动安装完成的情况下，按照第一篇文章的步骤，下载 Unlock Tool 并在网页上获取 OPT 口令。每天的 OPT 口令有上限，先到先得，并且每个口令有效期为 15 分钟。打开工具输入 OPT，并且使用 USB 连接处于 bootloader 模式下的手机，点击解锁按钮即可。

TWRP

成功解锁 bootloader 后，我们可以无限制的刷入三方固件了，首先要准备的是 TWRP，从网站上下载最新固件并命名为 twrp.img。
我首次尝试在 bootloader 模式中直接刷入固件：

1
fastboot flash recovery twrp.img

但是显示如下错误：

FAILED (remote: (recovery_b) No such partition)

于是我选择载入到 TWRP 中进行安装：

1
fastboot boot twrp.img

手机自动重启后，便进入了 TWRP 中，此时可以在选项中将当前 TWRP 覆盖手机的 recovery。

AB 分区

较新的手机现在都是有 AB 分区的，可以安装两个独立的系统，在 bootloader 模式下可以选择进入的分区：

1
fastboot set_active [a|b]

如果你在 A 分区安装了系统，但是激活了 B 分区，那么重启手机后将无法成功进入系统。同样，AB 分区也有两套 recovery，如果你只是将 TWRP 安装在 B 分区，那么 A 分区的 recovery 还是自带的。我将 TWRP 安装在了 B 分区，如果此时激活的是 A 分区，那么可以使用如下命令可以进入 B 分区的 recovery：

1
2
fastboot set_active b
fastboot reboot recovery

刷机

我选择的三方固件是 LineageOS，可以在论坛中下载镜像。在 TWRP 中，可以直接通过 USB 传输文件，也可以使用 adb 命令，能够很方便地将系统包传输到手机中。
在 TWRP 的界面中，先擦除掉所有数据，再进入安装界面并选中系统包开始刷入。如果此时激活的是 B 分区，那么系统的安装会默认选择 A 分区。安装完后重启之前，需要切换到 A 分区，然后重启。

分区

在首次刷入系统包时报错：

TWRP Error Applying Update: 28 kDownloadOperationExecutionError

搜索后得知是因为 Nokia 默认的 system 分区空间太小，需要重新分区，使用 Repartition tool 重新分区后重复上诉步骤即可。

Elkeid Golang RASP

简介

在以往的 RASP 解决方案中，部署方式通常需要业务参与，修改相关配置或是启动参数，这也就造成了 RASP 部署困难的窘境。更有甚者，由于 Golang 编译型语言的特性，多数 RASP 只能被迫选择在编译期集成进去，以降低技术实现成本，但这无疑又间接加大了部署推广的难度。
我始终认为限制 RASP 发展与推广的是部署，而并非是技术难度，许多厂商在各语言的技术实现上都有大同小异的成熟方案。例如使用 JVM 的 Instrumentation 功能动态修改 bytecode，可以在虚拟机功能的基础上实现稳定的运行时防护。
所以在 Elkeid RASP 的项目初期，团队就敲定了动态注入的部署方式，一切都朝着降低部署难度的目标靠拢。

原理

目前 Elkeid RASP 开源了 JVM、Python、Node 以及 Golang 四种语言的运行时保护功能，四种语言均支持对已存在的进程动态注入防护代码。其中 JVM、Node 依赖于虚拟机提供的机制，运行稳定，而 Python 与 Golang 则需要利用 ptrace 在进程层面上做注入。

进程注入

为了实现对 Python 以及 Golang 进程的动态防护，先不考虑运行时层面的代码篡改，我们至少需要一个能够在 Linux 任意进程空间内执行任意代码的工具。但是很可惜，Linux 没有类似于 CreateRemoteThread 的接口。
大多数的代码注入，都是使用 ptrace 篡改进程执行流程，调用 dlopen 加载动态库。而且大多数项目都会指出，该方式的不稳定性可能会导致进程永久卡住。因为 dlopen 底层会调用 malloc，而在 glibc 的官方文档中指明了 malloc 是不可重入函数。
在研究过程中，我发现了 mandibule 这个项目，它另辟蹊径地编写了一个 ELF Loader，再使用 ptrace 让该 ELF Loader 在目标进程内执行，加载一个全新的程序，执行完成后恢复主线程的寄存器。
由于作者已经放弃维护，而且我自己在使用过程中，发现了项目一些设计上的缺陷以及代码 bug。于是我借鉴了该思路，开发了 pangolin 这个工具，它可以在任意进程内临时运行另一个程序，细节可以看相关的 blog。

Inline Hook

借助 pangolin，我们可以在一个 Golang 的进程中执行任意代码，我们甚至可以篡改可执行段的机器码，很轻松地便可以对某个函数进行 Inline Hook。例如我们想对 Golang 的命令执行函数 exec.Command 进行 Inline Hook，那么可以在进程注入期间修改函数 os/exec.Command 的开头指令，使其执行时先跳转到我们编写的函数中。在我们自定义的函数中，便可以获取该函数调用时的入参以及调用栈，再通过某种通信方式传输出去，一个简单的 RASP 模型便完成了。
那么要完成该流程，我们需要一些先决条件：

  • 在去除掉 ELF 符号信息的情况下，如何获取 Golang 的符号信息，以确定函数的地址，完成对函数的 Inline Hook 操作。
  • Golang 如何进行函数调用，通过寄存器亦或是栈，我们又该如何读取函数的入参。
  • 如何获取 Golang 当前函数的 Stack Frame 长度，用于定位上一层函数的返回地址，完成调用栈回溯。

Golang Runtime Symbol Information

在去除了 ELF 符号信息后，一个编译好的 Golang 程序还是可以正确地执行 debug.PrintStack 函数，那么便可以证明 Golang 内部必然还存在一个符号表。根据官方文档 Go 1.2 Runtime Symbol Information 的介绍，Golang 从 1.2 之后的版本内置了符号信息，对于 ELF 格式来说，通常放置在 .gopclntab 这个 section。
这些符号信息不仅包含了函数名称、函数地址范围以及函数栈帧长度信息，甚至还有相关的代码文件名，以及行号等源码信息。根据文档记录的信息格式，我们可以很轻松地解析出一个 Golang 二进制程序的符号表。

Golang build info

对于 Golang 1.13 以上编译出的二进制，可以使用 go version 命令查看编译时的 Golang 版本，由此说明二进制中内嵌了相关的编译信息。对于 ELF 格式来说，编译信息存放在 .go.buildinfo section，其中包含 Golang 版本号以及三方依赖库列表。值得注意的是，buildinfo 的格式在 Golang 1.18 版本发生了改变，弃用了数据指针，相关解析代码可查看官方仓库。

Golang internal ABI

接下来我们需要了解 Golang 编译出的机器码，是如何进行函数调用的，以及 Golang 的结构体在内存中是如何存放的，了解这些之后我们才能正确地取出函数入参。

memory layout

Golang 包含的内置类型描述，可以在官方文档 The Go Programming Language Specification 中找到。对于数值类型，Golang 规范了类型的内存占用大小，而字节对齐则随 CPU 架构不同而变化。对于复合类型，例如 string、slice 以及 map 等，内存占用大小由组成的基础类型及其字节对齐决定，而该复合类型的字节对齐由组成类型中最大的字节对齐决定。
文档中并未描述 string 等内置类型的底层内存排布，但是我们可以从一些文章，亦或是 CGO 生成的头文件中一窥究竟。
以下是我使用 cpp 对 x64 架构下 Golang 类型的描述，代码可以在 Elkeid 官方仓库中找到：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
namespace go {
    typedef signed char Int8;
    typedef unsigned char Uint8;
    typedef short Int16;
    typedef unsigned short Uint16;
    typedef int Int32;
    typedef unsigned int Uint32;
    typedef long long Int64;
    typedef unsigned long long Uint64;
    typedef Int64 Int;
    typedef Uint64 Uint;
    typedef __SIZE_TYPE__ Uintptr;
    typedef float Float32;
    typedef double Float64;
    typedef float _Complex Complex64;
    typedef double _Complex Complex128;

    struct interface {
        void *t;
        void *v;
    };

    struct string {
        const char *data;
        ptrdiff_t length;
    };

    template<typename T>
    struct slice {
        T *values;
        Int count;
        Int capacity;
    };
}

可以看到 string 类型由两个字段组成，数据指针加上字符串长度，在内存中总共占用 16 字节。string 的内存对齐则由这两个字段决定，即 align(string) = max(align(const char *), align(ptrdiff_t))。
对于 int32 这些 Golang 的基础数值类型来说，其字节对齐与 cpp 默认的对齐一致。

Type 64-bit 32-bit
Size Align Size Align
bool, uint8, int8 1 1 1 1
uint16, int16 2 2 2 2
uint32, int32 4 4 4 4
uint64, int64 8 8 8 4
int, uint 8 8 4 4
float32 4 4 4 4
float64 8 8 8 4
complex64 8 4 8 4
complex128 16 8 16 4
uintptr, *T, unsafe.Pointer 8 8 4 4

但是 Golang 并不确保这些类型的字节对齐不变，官方似乎正在考虑改变 x86 上 int64 的字节对齐。现在我们已经了解了 Golang 类型的内存排布，那么对于任意入参的函数调用，我们都能准确的从内存中取出数据。现在剩下的问题便是函数调用发生时，参数将会存放在何处？

stack-based calling conventions

Golang 在 1.17 版本之前的函数调用中，参数与结果均存放在栈上。但由于栈上频繁的内存操作影响了运行性能，所以社区草拟了基于寄存器的调用约定方案，并在 1.17 版本后切换到该调用约定。
我们先了解 Golang 最原始的 ABI0，也就是基于栈的调用约定，细节描述可以从文档 A Quick Guide to Go’s Assembler 找到。在函数调用发生时，调用者需要将参数以及返回值，从低地址向高地址依次排列在栈顶。
例如在调用函数 func A(a int32, b string) (int32, error) 时，我们需要按下列排布存放参数与返回值：

+------------------------------+
| 2nd result error.v           |
| 2nd result error.t           |
| 1st result int32             |
| <pointer-sized alignment>    |
| b string.length              |
| b string.data                |
| a int32                      |
+------------------------------+ ↓ stack pointer

先放入 4 字节的参数 a，接着放入 string 类型的参数 b。由于 string 类型的字节对齐是 8，而此时的地址为 sp + 4，所以需要填充 4 字节的空白区域，从 sp + 8 开始放置 string 的数据。参数存放完成后，如果此时的地址没有按指针大小对齐，则需要填充空白字节。例如在 amd64 架构上，最后一个 int32 的参数放置于地址 0x40000，占用 4 字节大小，那么我们需要再填充 4 字节空白数据，使得返回值存放地址为 0x40008，按当前架构的指针大小 8 对齐。
我们接着放入第一个 int32 的返回值，而第二个返回值类型为 error，实际上就是 interface 类型。由上一小结可知，interface 类型占用 16 字节，按 8 字节对齐，所以我们填充 4 字节后，放入 error 结构体。当然，对于返回值而言，我们并不会真正地写入数据，而是预留内存空间以供被调用者写入。

register-based calling conventions

对于基于寄存器的调用约定，调用者需要先尝试将参数放置于寄存器中。如果结构体太大，或是结构体中包含 Non-trivial arrays 类型成员导致无法存放，则会转而放置于栈上。对于 amd64 架构，Golang 使用X0 – X14 寄存器存放浮点数数据，而对于整数数值，则使用以下 9 个整数寄存器存放：

RAX, RBX, RCX, RDI, RSI, R8, R9, R10, R11

对于数值类型参数，我们可以直接将参数一一对应到寄存器中。而对于结构体类型，我们需要将结构体拆解成多个基础数值类型，然后进行对应放置。如果一个结构体拆解后，需要占用的寄存器数超过了剩余的寄存器数，则该整个结构体都只能放置于栈上。该部分细节繁杂，本文不作赘述，细节请看文档 Go internal ABI specification。

实现

Runtime conflict

有了上述理论支持后，我们现在可以着手编写钩子函数了。在钩子函数执行过程中，不能随意篡改堆栈上的数据，执行完成后需要恢复所有寄存器，并跳转到原函数继续执行。需要注意的是，我们必须时刻记住，执行钩子函数的是 Golang 的线程，那么就存在以下两个问题：

  • Golang 为线程分配的栈空间很小，钩子函数如果使用过度会导致 Segmentation fault。
  • 在 Golang 的线程中执行时，我们无法正常调用 glibc 函数，例如 malloc 依赖于 fs 寄存器指向的 TLS 结构，以保证线程安全，但 fs 在 1.17 版本以下的 Golang 线程中指向全局 G。

为了解决第一个问题，我们需要在钩子函数的入口处，申请一块足够大的内存替换当前栈。而对于第二点，我们只能使用freestanding 代码及 syscall 来完成参数读取与栈回溯操作。为了更好地解耦与复用，于是我开发了一个不依赖于 glibc 的小型 c-runtime，包含内联汇编编写的 syscall 以及必要的标准库函数。我们可以安全地在 Golang 线程中调用 c-runtime 中的任何函数，例如使用底层是无锁环形缓冲区和 mmap syscall 的 z_malloc 来分配堆空间。

钩子函数

下面是使用内联汇编编写的钩子函数 wrapper，可以通用地进行栈替换、寄存器备份以及函数跳转：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
asm volatile(
    "mov $1, %%r12;"
    "mov %%rsp, %%r13;"
    "add $8, %%r13;"
    "and $15, %%r13;"
    "sub $16, %%rsp;"
    "movdqu %%xmm14, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm13, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm12, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm11, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm10, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm9, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm8, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm7, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm6, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm5, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm4, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm3, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm2, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm1, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm0, (%%rsp);"
    "push %%r11;"
    "push %%r10;"
    "push %%r9;"
    "push %%r8;"
    "push %%rsi;"
    "push %%rdi;"
    "push %%rcx;"
    "push %%rbx;"
    "push %%rax;"
    "sub %%r13, %%rsp;"
    "mov %0, %%rdi;"
    "call z_malloc;"
    "cmp $0, %%rax;"
    "je end_%=;"
    "mov %%rsp, %%rdi;"
    "mov %%rax, %%rsp;"
    "add %0, %%rsp;"
    "push %%rax;"
    "push %%rdi;"
    "add $312, %%rdi;"
    "add %%r13, %%rdi;"
    "call %P1;"
    "mov %%rax, %%r12;"
    "pop %%rsi;"
    "pop %%rdi;"
    "mov %%rsi, %%rsp;"
    "call z_free;"
    "end_%=:"
    "add %%r13, %%rsp;"
    "pop %%rax;"
    "pop %%rbx;"
    "pop %%rcx;"
    "pop %%rdi;"
    "pop %%rsi;"
    "pop %%r8;"
    "pop %%r9;"
    "pop %%r10;"
    "pop %%r11;"
    "movdqu (%%rsp), %%xmm0;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm1;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm2;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm3;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm4;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm5;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm6;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm7;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm8;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm9;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm10;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm11;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm12;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm13;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm14;"
    "add $16, %%rsp;"
    "cmp $0, %%r12;"
    "je block_%=;"
    "jmp *%2;"
    "block_%=:"
    "ret;"
    ::
    "i"(STACK_SIZE),
    "i"(handler),
    "m"(origin)
);

r12和r13 寄存器是 Golang 中可以随意使用的临时寄存器，我们用 r12 来标识是否要阻断当前调用。而r13 用来参与计算，以确保调用 handler 时栈指针按 16 字节对齐，这是 amd64 下 gcc 的默认约定。
在代码的开头，我们先将 X0 - X14 浮点数寄存器推入栈中，接着推入 Golang 1.17 以上需要使用的整数寄存器。然后调用 z_malloc 申请 40K 的内存替换当前栈，再以原始栈指针为参数调用 handler。在 handler 函数中，根据 Golang 的版本不同，我们可以从栈上存储的寄存器中，或上一函数的 Stack frame 中读出入参。当然也可以根据原始栈指针读取返回地址，从 Golang 符号表中查找函数信息，然后根据 Stack frame 读出上一层的返回地址，循环往复完成栈回溯。
我们甚至可以在 handler 中判断参数是否合法，当参数匹配到我们设置的正则时，可以手动写入 error 返回值到栈上，并返回 false 以将 r12 寄存器置零完成阻断。在 handler 函数执行完成后，从栈上恢复寄存器，并根据 r12 决定返回还是跳转至原函数。
为了更好地进行参数读取和阻断，我使用Templates 编写了一套 Golang 类型反射库，可以在运行时获取 Golang 类型元数据。元数据包含类型的基础类型成员数，每个成员的相对偏移以及占用大小，还有该类型需要占用的浮点/整数寄存器数。在 handler 函数中，我们可以轻松地利用这些元数据分析 Golang 的参数内存布局，正确地取出数据。由于该部分代码细节繁多，限于本文篇幅所以不进行详细讲解，取参与回溯部分请直接阅读仓库代码。

回溯停止

对于调用栈的回溯，上面已经解析过了，我们可以取出当前栈顶的返回地址，在符号表中查找地址相关的函数的名称、文件、行号以及栈帧大小。获取栈帧大小后，取出 sp + framesize 的上一层返回地址，循环上述步骤即可。但有一个问题是，我们在哪里结束循环？调用链的层数一定有限，那么第一个函数是哪个？
在 1.2 版本中，Golang 通过判断函数名是否为 runtime.goexit、runtime.rt0_go 等入口函数，由此决定是否终止回溯。而对于较新版本的 Golang ，符号信息中增加了一个 funcID 字段，通过 funcID 判断函数类型是否为入口函数。但 funcID 的本质与函数名比较无二，而且 funcID 在版本之间会发生变动，所以最后决定简单地使用函数名判断：

1
2
3
4
5
6
7
8
9
10
constexpr auto STACK_TOP_FUNCTION = {
        "runtime.mstart",
        "runtime.rt0_go",
        "runtime.mcall",
        "runtime.morestack",
        "runtime.lessstack",
        "runtime.asmcgocall",
        "runtime.externalthreadhandler",
        "runtime.goexit"
};

消息通信

成功获取入参和调用栈后，要如何把消息传输出去？如果需要进行 socket 通信，并且不阻塞 Golang 线程，那就需要驻留一个线程在 Golang 进程内，实现一个简单的生产者消费者模型。那么在无法使用 std::queue 等标准库的情况下，要怎么实现消费丢列，又该如何保证线程安全？
为了尽可能地减少性能影响，我利用 gcc 内置的原子操作实现了一个定长的无锁环形缓冲区，并使用 c-runtime 中实现的 condition variable 做线程同步，实现了一个高效的消息队列。在每个钩子函数触发时，都会将入参和调用栈打包放入队列，如果队列已满则丢弃该消息。
在 pangolin 注入过程中，我们启动一个消费者线程，从消息队列中消费函数调用信息，序列化为 json 后通过 unix socket 传输到 server。

信号屏蔽

Golang 启动时会设置信号处理函数，而在进程收到信号时，内核会随机选择一个线程进行信号处理。我们在 Golang 进程中驻留的几个 cpp 线程有可能被选中用于执行处信号处理函数，但是处理函数默认当前处于 Golang 线程中，读取 fs 寄存器以访问 Golang 的全局 G，但此时 fs 所指向的其实是 glibc 的 TLS，于是导致异常退出。
为了避免这种情况发生，我们需要手动设置驻留的 cpp 线程，令其屏蔽所有信号：

1
2
3
4
5
6
7
8
9
sigset_t mask = {}; 
sigset_t origin_mask = {};

sigfillset(&mask);

if (pthread_sigmask(SIG_SETMASK, &mask, &origin_mask) != 0) {
    LOG_ERROR("set signal mask failed");
    quit(-1);
}

流程

在学习了原理和实现细节后，我们来解析一下 go-probe 的执行流程，入口函数如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
#include "go/symbol/build_info.h"
#include "go/symbol/line_table.h"
#include "go/symbol/interface_table.h"
#include "go/api/api.h"
#include <zero/log.h>
#include <csignal>
#include <asm/api_hook.h>
#include <z_syscall.h>

void quit(int status) {
    uintptr_t address = 0;
    char *env = getenv("QUIT");

    if (!env) {
        LOG_WARNING("can't found quit env variable");
        z_exit_group(-1);
    }

    if (!zero::strings::toNumber(env, address, 16) || !address) {
        LOG_ERROR("invalid quit function address");
        z_exit_group(-1);
    }

    ((decltype(quit) *)address)(status);
}

int main() {
    INIT_FILE_LOG(zero::INFO, "go-probe");

    sigset_t mask = {};
    sigset_t origin_mask = {};

    sigfillset(&mask);

    if (pthread_sigmask(SIG_SETMASK, &mask, &origin_mask) != 0) {
        LOG_ERROR("set signal mask failed");
        quit(-1);
    }

    if (!gLineTable->load()) {
        LOG_ERROR("line table load failed");
        quit(-1);
    }

    if (gBuildInfo->load()) {
        LOG_INFO("go version: %s", gBuildInfo->mVersion.c_str());

        CInterfaceTable table = {};

        if (!table.load()) {
            LOG_ERROR("interface table load failed");
            quit(-1);
        }

        table.findByFuncName("errors.(*errorString).Error", (go::interface_item **)CAPIBase::errorInterface());
    }

    gSmithProbe->start();

    for (const auto &api : GOLANG_API) {
        for (unsigned int i = 0; i < gLineTable->mFuncNum; i++) {
            CFunc func = {};

            if (!gLineTable->getFunc(i, func))
                break;

            const char *name = func.getName();
            void *entry = (void *)func.getEntry();

            if ((api.ignoreCase ? strcasecmp(api.name, name) : strcmp(api.name, name)) == 0) {
                LOG_INFO("hook %s: %p", name, entry);

                if (hookAPI(entry, (void *)api.metadata.entry, api.metadata.origin) < 0) {
                    LOG_WARNING("hook %s failed", name);
                    break;
                }

                break;
            }
        }
    }

    pthread_sigmask(SIG_SETMASK, &origin_mask, nullptr);
    quit(0);

    return 0;
}

需要明确的是，go-probe 由 pangolin 注入到 Golang 进程的主线程中临时运行。同时 pangolin 使用 ptrace 持续监听该线程的 syscall 调用，拦截到 main 函数发出的 exit 或 exit_group 调用后，恢复线程状态并结束注入流程。然而可以看到，上面的代码中会优先调用环境变量 QUIT 指向的函数，这又是为何？
在实际的部署过程中，由于资源限制等诸多特殊原因，pangolin 进程可能会在注入期间被 kill。那么此时 main 函数执行的 syscall 就无人拦截，exit_group 会真正地导致业务进程退出。为了让执行 go-probe 的线程能够自我恢复，pangolin 会提前将线程状态快照写入到 Golang 内存中。同时遗留在 Golang 进程中的 shellcode 包含一个 quit 函数，能够根据该快照主动恢复线程，类似于 glibc 的 setcontext，而 QUIT 环境变量正是 quit 的地址。
在初始化文件日志后，先令当前线程屏蔽所有信号，之后启动的所有子线程都会继承该设置。然后从 ELF 的 section 中加载符号表、编译信息以及 interface 表，并且为了支持阻断功能，查找 errors.(*errorString) 的地址并保存。执行 gSmithProbe->start() 启动通信客户端后，从符号表中查找 GOLANG_API 所有子项，并进行 Inline Hook。完成以上流程后，恢复信号掩码并调用 quit 函数以通知 pangolin 结束注入。

Node.js进程注入

前言

上篇文章实现了对Python进程的代码注入，今天接着来尝试Node.js的注入。

Inspector

根据Node.js的官方文档得知，在Node.js 8之后加入了新的调试机制(Inspector API)，在启动时加入”–inspect”参数或者向进程发送SIGUSR1信号，就会激活检查器。

1
2
$ node
$ kill -USR1 $(pidof node)

激活检查器后，终端会打印信息：

Debugger listening on ws://127.0.0.1:9229/f1f12ee0-4f35-4e61-836c-71d175a607e3

检查器监听本地端口9929，使用websocket协议，协议约定可见文档，连接检查器可以使用node自带的inspect client：

1
$ node inspect -p $(pidof node)

连接后可以通过”exec(‘console.log(123)’)”在目标进程中执行node代码，或者执行”repl”开启交互模式。但是自带的inspect client功能简单，如果需要更好的体验，可以使用chrome的调试模块，在chrome中访问”chrome://inspect”即可。

协议

node-inspect就是检查器客户端的官方代码仓库，通过阅读inspect_client.js发现，交互协议就是简单的json字符串，每个请求和响应通过id字段一一对应。
例如请求开启调试时，通过websocket协议发送字符串：

1
2
3
4
{
    "id": 1,
    "method": "Debugger.enable"
}

检查器返回响应：

1
2
3
4
5
6
{
    "id": 1,
    "result": {
        "debuggerId": "(D750AD89818A150E3155AC1D6ECB7BB)"
    }
}

如果你想在目标进程中执行代码，类似于”exec(‘console.log(123)’)”命令，你只需要使用Runtime.evaluate，发送请求：

1
2
3
4
5
6
7
8
{
    "id": 2,
    "method": "Runtime.evaluate",
    "params": {
        "expression": "console.log(123)",
        "includeCommandLineAPI": true
    }
}

参数需要设置”includeCommandLineAPI”，否则注入的代码无法使用require函数。

Python进程注入

前言

上篇文章讨论了Linux进程注入，能够在机器码层面进行代码注入，但如果现在有一个Python进程在运行，如何才让将Python代码注入进去呢？

inject by gdb

目前有两个项目能够对Python虚拟机进行注入，分别是pylane以及pyrasite。简单浏览两个项目，可以发现都是使用GDB在进程中进行函数调用：

1
2
3
4
5
6
7
8
9
10
11
12
13
class Injector(object):
    def generate_gdb_codes(self):
        # ...
        return [
            # use char in case of symbol PyGilState_STATE not found
            'call $gil_state = (char) PyGILState_Ensure()',
            'call (void) PyRun_SimpleString("%s")' % prepare_code,
            'call (void) PyRun_SimpleString("%s")' % run_code,
            'call (void) PyRun_SimpleString("%s")' % cleanup_code,
            # make sure previous codes are safe.
            # gdb exit without GIL release is a disaster for target process.
            'call (void) PyGILState_Release($gil_state)',
        ]

GDB附加到进程后，使用call命令在目标进程中依次调用函数PyGILState_Ensure、PyRun_SimpleString、PyGILState_Release就能在Python虚拟机中运行代码。

1
2
3
4
5
6
7
sudo gdb \
  -p $pid \
  -ex 'call (int)PyGILState_Ensure()' \
  -ex 'call (int)PyRun_SimpleString("print(123)")'  \
  -ex 'call (int)PyGILState_Release($1)' \
  -ex 'set confirm off' \
  -ex quit

使用上诉命令，可以直接将”print(123)”注入到Python进程中。

inject by pangolin

既然只需要在进程中调用三个函数，就可以完成Python代码注入，那么我们是否可以使用pangolin将某个程序注入到进程中，完成这三个函数的调用。这也就是项目python-inject的由来，它使用pangolin进行Python代码注入，不依赖于GDB。

符号查找

首先我们需要查找函数在目标进程中的地址，解析Python的ELF符号表，然后加上进程基址，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
#include <common/log.h>
#include <common/cmdline.h>
#include <common/utils/process.h>
#include <elfio/elfio.hpp>

typedef int (*PFN_RUN)(const char *command);
typedef int (*PFN_ENSURE)();
typedef void (*PFN_RELEASE)(int);

constexpr auto PYTHON = "bin/python";
constexpr auto PYTHON_LIBRARY = "libpython";
constexpr auto PYTHON_CALLER = "python_caller";

int main(int argc, char ** argv) {
    cmdline::parser parse;

    parse.add<int>("pid", 'p', "pid", true, 0);
    parse.add<std::string>("source", 's', "python source file", true, "");
    parse.add<std::string>("pangolin", '\0', "pangolin path", true, "");
    parse.add("file", '\0', "pass source by file");

    parse.parse_check(argc, argv);

    int pid = parse.get<int>("pid");

    CProcessMap processMap;

    if (
            !CProcess::getFileMemoryBase(pid, PYTHON_LIBRARY, processMap) &&
            !CProcess::getFileMemoryBase(pid, PYTHON, processMap)
            ) {
        LOG_ERROR("find target failed");
        return -1;
    }

    LOG_INFO("find target: 0x%lx -> %s", processMap.start, processMap.file.c_str());

    ELFIO::elfio reader;

    if (!reader.load(processMap.file)) {
        LOG_ERROR("open elf failed: %s", processMap.file.c_str());
        return -1;
    }

    auto it = std::find_if(
            reader.sections.begin(),
            reader.sections.end(),
            [](const auto& s) {
                return s->get_type() == SHT_DYNSYM;
            });

    if (it == reader.sections.end()) {
        LOG_ERROR("can't find symbol section");
        return -1;
    }

    unsigned long baseAddress = 0;

    if (reader.get_type() != ET_EXEC) {
        auto sit = std::find_if(
                reader.segments.begin(),
                reader.segments.end(),
                [](const auto& s) {
                    return s->get_type() == PT_LOAD;
                });

        if (sit == reader.segments.end()) {
            LOG_ERROR("can't find load segment");
            return -1;
        }

        baseAddress = processMap.start - (*sit)->get_virtual_address();
    }

    PFN_ENSURE pfnEnsure = nullptr;
    PFN_RUN pfnRun = nullptr;
    PFN_RELEASE pfnRelease = nullptr;

    ELFIO::symbol_section_accessor symbols(reader, *it);

    for (ELFIO::Elf_Xword i = 0; i < symbols.get_symbols_num(); i++) {
        std::string name;
        ELFIO::Elf64_Addr value = 0;
        ELFIO::Elf_Xword size = 0;
        unsigned char bind = 0;
        unsigned char type = 0;
        ELFIO::Elf_Half section = 0;
        unsigned char other = 0;

        if (!symbols.get_symbol(i, name, value, size, bind, type, section, other)) {
            LOG_ERROR("get symbol %lu failed", i);
            return -1;
        }

        if (name == "PyGILState_Ensure")
            pfnEnsure = (PFN_ENSURE)(baseAddress + value);
        else if (name == "PyRun_SimpleString")
            pfnRun = (PFN_RUN)(baseAddress + value);
        else if (name == "PyGILState_Release")
            pfnRelease = (PFN_RELEASE)(baseAddress + value);
    }

    if (!pfnEnsure || !pfnRun || !pfnRelease) {
        LOG_ERROR("can't find python symbols");
        return -1;
    }

    LOG_INFO("ensure func: %p run func: %p release func: %p", pfnEnsure, pfnRun, pfnRelease);

    std::string source = parse.get<std::string>("source");
    std::string pangolin = parse.get<std::string>("pangolin");
    std::string caller = CPath::join(CPath::getAPPDir(), PYTHON_CALLER);

    char callerCommand[1024] = {};

    snprintf(
            callerCommand, sizeof(callerCommand),
            "%s %s %d %p %p %p",
            caller.c_str(), source.c_str(), parse.exist("file"),
            pfnEnsure, pfnRun, pfnRelease
            );

    int err = execl(
            pangolin.c_str(), pangolin.c_str(),
            "-c", callerCommand,
            "-p", std::to_string(pid).c_str(),
            nullptr
            );

    if (err < 0) {
        LOG_ERROR("exec pangolin failed: %s", strerror(errno));
        return -1;
    }

    return 0;
}

函数调用

我们现在知道了函数在进程中的地址，只需要注入程序并传递地址参数，在进程中完成三个函数的调用，就可以实现Python代码注入：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
#include <crt_asm.h>
#include <crt_log.h>
#include <crt_utils.h>

typedef int (*PFN_RUN)(const char *command);
typedef int (*PFN_ENSURE)();
typedef void (*PFN_RELEASE)(int);

int main(int argc, char ** argv, char ** env) {
    if (argc < 6) {
        LOG("usage: ./program source [0|1](source is file) address address address");
        return -1;
    }

    char *source = argv[1];
    int is_file = !strcmp(argv[2], "1");

    PFN_ENSURE pfn_ensure = (PFN_ENSURE)strtoul(argv[3], NULL, 16);
    PFN_RUN pfn_run = (PFN_RUN)strtoul(argv[4], NULL, 16);
    PFN_RELEASE pfn_release = (PFN_RELEASE)strtoul(argv[5], NULL, 16);

    if (!pfn_ensure || !pfn_run || !pfn_release) {
        LOG("func address error");
        return -1;
    }

    LOG("inject python source: %s flag: %d", source, is_file);

    char *code = source;

    if (is_file) {
        char *buffer = NULL;
        long length = read_file(source, &buffer);

        if (length <= 0) {
            LOG("read file failed: %s", source);
            return -1;
        }

        code = malloc(length + 1);

        if (!code) {
            LOG("malloc code memory failed");
            free(buffer);
            return -1;
        }

        memset(code, 0, length + 1);
        memcpy(code, buffer, length);

        free(buffer);
    }

    int state = pfn_ensure();
    int err = pfn_run(code);
    pfn_release(state);

    if (is_file) {
        free(code);
    }

    return err;
}

void _main(unsigned long * sp) {
    int argc = *sp;
    char **argv = (char **)(sp + 1);
    char **env = argv + argc + 1;

    __exit(main(argc, argv, env));
}

void _start() {
    CALL_SP(_main);
}

fs寄存器

也许你有疑问，为什么要将函数计算和调用分开，为什么不直接在目标进程中计算函数地址？实际上，我在项目刚开始的时候也是这样设计的，但是在测试过程中一直crash，经过调试发现这和fs寄存器相关。

glibc

如果你使用C/C++编写了一个依赖于标准库的程序，那么程序加载时会先由glibc进行初始化，而在glibc中，每个线程初始化时都会修改fs寄存器指向一段私有的内存，用来存储线程局部变量等数据。
Python也会设置私有的fs寄存器，如果将一个依赖glibc的程序注入进去，那么程序运行时会修改fs寄存器。而此时调用Python的函数，就会导致crash，因为函数内部会访问fs寄存器，但此时的fs寄存器已经被修改了。
为了避免fs寄存器被修改，只能编写一个不依赖于标准库的程序(-nostdlib)，类似于我上面的代码，从程序入口点”_start”开始编写。不使用标准库进行符号解析以及地址计算，开发难度较大，所以权衡之下我选择将两者分离。

arch_prctl

使用arch_prctl系统调用可以修改fs寄存器，所以我尝试过一种解决方法，在函数调用之前临时将fs寄存器修改为原有值。所以我修改了pangolin，在进行注入时先获取当前fs寄存器值存入环境变量，在程序注入后根据环境变量恢复寄存器。
但是结果还是会导致crash，因为glibc的架构复杂，所以我没有继续深究，总之之后涉及fs寄存器的一切我都会敬而远之。

Linux进程注入

前言

将代码注入到特定进程空间内执行，是一种十分geek的行为，可用于破解、安全防护等。不像Windows有CreateRemoteThread可以直接在目标进程内创建线程，在Linux下进行进程注入会显得复杂些。
目前比较常用的手段，就是计算dlopen函数在libc.so中的位置，然后利用ptrace连接目标进程，更改PC地址使其运行该函数，但是该方式只能针对非静态编译的程序。

mandibule

我在github上发现了项目mandibule，该项目不使用任何C标准库(-nostdlib)，类似于strcmp之类的函数都用C重写一遍，所以编译出来的产物就是一个无依赖shellcode。

虚拟地址空间

作者巧妙地使用编译器默认不进行函数排序的特点，在唯一的源文件mandibule.c开头编写函数：

1
2
3
4
5
6
unsigned long mandibule_beg(int aligned)
{
    if(!aligned)
        return (unsigned long)mandibule_beg;
    return (unsigned long)mandibule_beg - ((unsigned long)mandibule_beg % 0x1000);
}

因为项目只有这一个源文件，所以编译出来只有一个object文件，也就是说该函数会在程序代码段的最开头。在mandibule.c的末尾编写函数：

1
2
3
4
5
6
unsigned long mandibule_end(void)
{
    uint8_t * p = (uint8_t*)"-= end_rodata =-";
    p += 0x1000 - ((unsigned long)p % 0x1000);
    return (unsigned long)p;
}

编译后该函数当然会在代码段的末尾，作者编写这两个函数的目的是为了在该程序运行时，可以通过调用mandibule_beg(1)、mandibule_end获得该程序的虚拟地址范围。
一般来说ELF加载到内存中时，ELF头部之后就是TEXT代码段，又因为这些段是按页对齐的，所以调用mandibule_beg(1)返回自身函数地址向下对齐页，大概率获得ELF头部的虚拟地址，也就是程序虚拟空间的开始。
但是获取程序虚拟空间末尾地址却没这么简单，因为在TEXT段后面还有DATA、RODATA等数据段。mandibule_end内使用了一个用不到字符串，该字符串肯定会在RODATA中，所以使用该字符串地址向上对齐页能够获得末尾的虚拟地址。
根据这两个地址，我们可以确定程序运行所需的代码、数据地址空间，你甚至可以将该范围的数据全部拷贝到一个堆空间，更改页属性为可执行，然后直接跳转到入口点运行起来，当然，这需要你编译的代码是位置无关的(-PIC)。

注入流程

mandibule进行代码注入的步骤：

  • 使用ptrace附加到目标进程
  • 备份进程寄存器、内存数据
  • 将自身运行时地址空间内的所有数据拷贝到目标进程内
  • 修改目标进程的寄存器，其实跳转到入口地址
  • 等待目标进程调用exit系统调用
  • 恢复寄存器、内存数据
  • 恢复进程的运行

第三步用到的地址范围就是mandibule_beg、mandibule_end所获得的，也相当于将程序自身拷贝到目标进程中运行。程序在目标进程中运行时，会使用一个自己编写ELF loader加载需要注入的代码。主要是使用mmap将程序段映射到进程空间，如果依赖动态库则还需要映射interpreter，最后生成一个假栈并设置好argv、env等信息，最后直接跳转到程序入口地址。

缺陷

使用mandibule可以对静态编译的程序进行注入，但是项目的设计有些缺陷：

  • 没有将shellcode与ptrace注入剥离，导致耦合严重
  • 项目不使用C标准库，进行二次开发较为麻烦

另外我在使用中发现了许多的bug，可见项目issue，我在推特上联系了作者，但他并没有要接着进行维护的意思。

pangolin

最后我决定对项目进行重构，将ELF loader编写成独立的shellcode模块，而ptrace注入部分完全可以用熟悉的C++开发。具体的代码细节不赘述，可以自行在github上查看项目。

链接脚本

在我的设计中，将shellcode部分编译成独立的so文件，然后根据导出符号的地址获取所有shellcode数据，将其写入目标进程中运行。但是在实现中遇到了问题，因为代码编译后不仅仅只有代码段，还包括RODATA等数据段，我们需要将所有数据都包括进来，而导出符号只会出现在TEXT中。
我知道在链接阶段可以指定所有符号的地址，让编译器根据我们的要求进行符号分段、排序。最后通过搜索学习，我编写了链接脚本：

1
2
3
4
5
6
7
8
9
10
SECTIONS
{
    .text :
    {
        *(.begin);
        *(.text.*);
        *(.rodata.*);
        *(.end);
    }
}

使用该脚本进行链接，最后生成的程序只会有一个TEXT段，字符串等数据都会在该段中。另外我编写了三个导出函数：

1
2
3
void __attribute__ ((section (".begin"))) shellcode_begin();
void shellcode_start();
void __attribute__ ((section (".end"))) shellcode_end();

可以看到我指定了两个section分配给shellcode_begin、shellcode_end，也对应了链接脚本中的begin、end。所以最后生成的程序中，符号shellcode_begin会在TEXT段的最开始，shellcode_end在TEXT的末尾。
我们只需要将shellcode_begin至shellcode_end之间的数据拷贝到目标进程空间，然后跳转到shellcode_start - shellcode_begin的入口偏移处，就可以在进程中运行我们编写的shellcode。

多线程

mandibule使用ptrace附加到进程主线程，此时别的线程依旧在运行中，在将数据拷贝到目标进程中时，覆盖的范围一般从ELF头部开始。如果覆盖的范围较大，就可能覆盖到代码段，而恰好此时其余的线程调用了一个被覆盖了的函数，就会导致程序崩溃。
为了减少多线程进程crash的可能性，我对内存覆盖进行了优化，先覆盖一小段shellcode，在目标进程中调用mmap系统调用申请内存，用申请的内存来储存较大的ELF loader。并且在ptrace阶段，会附加到所有线程，使其临时暂停运行。

栈空间

shellcode在目标进程中运行时，使用的是线程暂停时的栈，对于普通的程序，栈空间足够，所以不需要担心出现意外情况。但是对于golang这类自己管理栈的程序，栈空间十分小，shellcode在进程中运行时可能会溢出栈空间，覆盖掉程序数据最终导致crash。
另外，如果想在目标进程中驻留，可以在注入后创建新线程，但是在程序恢复运行后，新线程调用getenv等API运行会导致crash。因为argv、env等数据存在于我们注入时构建的栈上，但是该栈是别的线程拥有的，可能数据早已丢失。
为了解决这两个问题，我在ELF loader的开头使用mmap申请一块足够大的栈空间，并且修改SP寄存器更换栈空间。

1
2
3
4
5
6
7
8
void loader_main(void *ptr) {
    LOG("elf loader start");

    char *stack = malloc(STACK_SIZE);
    char *stack_top = stack + STACK_SIZE;

    FIX_SP_JMP(stack_top, elf_loader, ptr);
}

还有许多的小细节，都存在于代码中，如果感兴趣可以自行阅读。

flutter逆向工程一

前言

几个星期前, 出于某些原因我想逆向某个APK文件, 和平常一样, 我使用apktool/dex2jar/jd-gui三件套开始工作. 出乎意料的是, 我没有在classes.dex里面找到任何程序相关的逻辑, 取而代之的是一些”flutter”的代码段.

初探

flutter是谷歌开发的跨平台框架, 初略的探索之后, 我得知程序的逻辑都存在于”libapp.so”文件中.

1
2
3
4
5
6
7
8
9
10
11
12
13
➜ tree lib
lib
├── arm64-v8a
│   ├── libapp.so
│   └── libflutter.so
├── armeabi-v7a
│   ├── libapp.so
│   └── libflutter.so
└── x86_64
    ├── libapp.so
    └── libflutter.so

3 directories, 6 files

APK解开后, lib目录下面会有两个库文件, “libflutter.so”是框架的运行时支持, “libapp.so”则是逻辑层编译而成的产物.

1
2
3
4
5
6
7
8
9
10
➜ readelf --symbols --wide libapp.so

Symbol table '.dynsym' contains 6 entries:
   Num:    Value  Size Type    Bind   Vis      Ndx Name
     0: 00000000     0 NOTYPE  LOCAL  DEFAULT  UND
     1: 00001000    12 FUNC    GLOBAL DEFAULT    1 _kDartBSSData
     2: 00002000  9616 FUNC    GLOBAL DEFAULT    2 _kDartVmSnapshotInstructions
     3: 00005000 24400 FUNC    GLOBAL DEFAULT    3 _kDartVmSnapshotData
     4: 0000b000 0x1edf30 FUNC    GLOBAL DEFAULT    4 _kDartIsolateSnapshotInstructions
     5: 001f9000 0x175bb0 FUNC    GLOBAL DEFAULT    5 _kDartIsolateSnapshotData

查看导出符号, 只有上面四个, 但从名称上分析并不像函数. 接下来开始使用IDA分析:

符号”_kDartVmSnapshotData”看起来的确不像是导出函数, 更像是序列化之后的数据, 应该是提供给dart虚拟机解释执行的.

探索

我搜索了很久, 几乎没有关于flutter逆向相关的文章, 只有reverse-engineering-flutter-apps-part-1这篇文章进行了剖析.

快照

“libapp.so”实际上是dart编译生成的快照文件, 这种快照格式称为”AOT”, 上面提到的四个导出符号就是快照数据的起始偏移:

  • “_kDartVmSnapshotData”: 虚拟机运行所需的Object等相关数据
  • “_kDartVmSnapshotInstructions”: 虚拟机运行所需指令相关数据
  • “_kDartIsolateSnapshotData”: 逻辑层运行所需的Object等相关数据
  • “_kDartIsolateSnapshotInstructions”: 逻辑层运行所需指令相关数据

官方没有文档公布快照AOT快照的数据格式, 不过幸好flutter和dart都是开源的, 我们可以调试AOT生成/加载的全过程, 就能知道快照里存了什么数据, 在逆向过程中又能如何利用这些数据.

编译

文章中提到了如何单纯的将dart文件转变成”libapp.so”:

1
2
~/flutter/bin/cache/dart-sdk/bin/dart ~/flutter/bin/cache/artifacts/engine/linux-x64/frontend_server.dart.snapshot --sdk-root ~/flutter/bin/cache/artifacts/engine/common/flutter_patched_sdk_product/ --strong --target=flutter --aot --tfa -Ddart.vm.product=true --output-dill app.dill main.dart
gen_snapshot --causal_async_stacks --deterministic --snapshot_kind=app-aot-elf --elf=libapp.so --strip app.dill

上面使用的是flutter sdk中包含的dart相关工具, 我测试之后发现只需要dart sdk就可以完成AOT的生成. 那么接下来, 我们需要编译一个带有调试信息的dart sdk:

1
2
3
4
5
6
7
8
9
10
11
# dev tool
git clone https://chromium.googlesource.com/chromium/tools/depot_tools.git
export PATH="$PATH:$PWD/depot_tools"

# fetch source
mkdir dart-sdk; cd dart-sdk
fetch dart

# build
cd sdk
./tools/build.py -m debug --no-goma -a x64 --debug-opt-level 0 all

查看编译生成的产物:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
➜ tree -L 1
.
├──  -
├── app.dill
├── args.gn
├── build.ninja
├── build.ninja.d
├── compressed_observatory_archive.tar
├── dart
├── dart2js_nnbd_strong_outline.dill
├── dart2js_nnbd_strong_platform.dill
├── dart2js_nnbd_strong_platform.dill.d
├── dart2js_outline.dill
├── dart2js_platform.dill
├── dart2js_platform.dill.d
├── dart2js_server_nnbd_strong_outline.dill
├── dart2js_server_nnbd_strong_platform.dill
├── dart2js_server_nnbd_strong_platform.dill.d
├── dart2js_server_outline.dill
├── dart2js_server_platform.dill
├── dart2js_server_platform.dill.d
├── dartdevc.dill
├── dartdevc.js
├── dartdevc.js.map
├── dartdev.dill
├── dart_precompiled_runtime
├── dart_precompiled_runtime_product
├── dart-sdk
├── ddc_outline.dill
├── ddc_outline_sound.dill
├── ddc_platform.dill
├── ddc_platform.dill.d
├── ddc_platform_sound.dill
├── ddc_platform_sound.dill.d
├── dds.dart.snapshot
├── dev_compiler
├── exe.stripped
├── gen
├── gen_kernel_bytecode.dill
├── gen_snapshot
├── gen_snapshot_fuchsia
├── gen_snapshot_host_targeting_host
├── gen_snapshot_product
├── gen_snapshot_product_fuchsia
├── gen_snapshot_product_host_targeting_host
├── icudtl.dat
├── icudtl_extra.dat
├── kernel-service.dart.snapshot
├── libapp.so
├── libentrypoints_verification_test_extension.so
├── libentrypoints_verification_test_extension.so.TOC
├── libffi_test_dynamic_library.so
├── libffi_test_dynamic_library.so.TOC
├── libffi_test_functions.so
├── libffi_test_functions.so.TOC
├── libsample_extension.so
├── libsample_extension.so.TOC
├── libtest_extension.so
├── libtest_extension.so.TOC
├── obj
├── observatory_archive.tar
├── offsets_extractor
├── offsets_extractor_precompiled_runtime
├── out.js
├── out.js.deps
├── out.js.map
├── process_test
├── run_vm_tests
├── toolchain.ninja
├── vm_outline_strong.dill
├── vm_outline_strong_product.dill
├── vm_outline_strong_stripped.dill
├── vm_platform_strong.dill
├── vm_platform_strong.dill.d
├── vm_platform_strong.dill.S
├── vm_platform_strong_product.dill
├── vm_platform_strong_product.dill.d
├── vm_platform_strong_stripped.dill
├── vm_platform_strong_stripped.dill.d
└── zlib_bench

5 directories, 73 files

dart-sdk子目录就和我们在官网下载到的版本一样, 但是它里面的程序符号表已被去除, 我们无法基于源码调试, 我们只能使用顶层那些散乱的产物. 根据官网的描述, 使用dart2native就能生成AOT快照:

1
dart2native bin/main.dart -k aot

在我们的目标产物中, 查看dart2native内容, 发现它只是个脚本文件:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
#!/usr/bin/env bash
# Copyright (c) 2019, the Dart project authors.  Please see the AUTHORS file
# for details. All rights reserved. Use of this source code is governed by a
# BSD-style license that can be found in the LICENSE file.

# Run dart2native.dart.snapshot on the Dart VM

function follow_links() {
  file="$1"
  while [ -h "$file" ]; do
    # On Mac OS, readlink -f doesn't work.
    file="$(readlink "$file")"
  done
  echo "$file"
}

# Unlike $0, $BASH_SOURCE points to the absolute path of this file.
PROG_NAME="$(follow_links "$BASH_SOURCE")"

# Handle the case where dart-sdk/bin has been symlinked to.
BIN_DIR="$(cd "${PROG_NAME%/*}" ; pwd -P)"
SNAPSHOTS_DIR="${BIN_DIR}/snapshots"
DART="$BIN_DIR/dart"

exec "$DART" "${SNAPSHOTS_DIR}/dart2native.dart.snapshot" $*

脚本使用dart执行dart2native.dart.snapshot, 说明dart2native.dart.snapshot是由dart编写而成的, 具体的源码可以查看github仓库. 通过阅读源码, 发现它其实是通过调用gen_snapshot进行的快照生成:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
final String kernelFile = path.join(tempDir.path, 'kernel.dill');
final kernelResult = await generateAotKernel(Platform.executable, genKernel,
    productPlatformDill, sourcePath, kernelFile, packages, defines,
    enableExperiment: enableExperiment);
if (kernelResult.exitCode != 0) {
  stderr.writeln(kernelResult.stdout);
  stderr.writeln(kernelResult.stderr);
  await stderr.flush();
  throw 'Generating AOT kernel dill failed!';
}

if (verbose) {
  print('Generating AOT snapshot.');
}

final String snapshotFile = (outputKind == Kind.aot
    ? outputPath
    : path.join(tempDir.path, 'snapshot.aot'));
final snapshotResult = await generateAotSnapshot(genSnapshot, kernelFile,
    snapshotFile, debugPath, enableAsserts, extraOptions);
if (snapshotResult.exitCode != 0) {
  stderr.writeln(snapshotResult.stdout);
  stderr.writeln(snapshotResult.stderr);
  await stderr.flush();
  throw 'Generating AOT snapshot failed!';
}

先使用dart生成kernel.dill, 然后使用gen_snapshot将kernel.dill转成AOT, 这也与文章中提到的方法一致.

生成

进入dart-sdk子目录, 编辑hello.dart:

1
2
3
void main() {
  print('Hello, World!');
}

接着生成kernel.dill:

1
./bin/dart bin/snapshots/gen_kernel.dart.snapshot --platform lib/_internal/vm_platform_strong_product.dill --aot -Ddart.vm.product=true -o kernel.dill hello.dart

使用gen_snapshot生成AOT快照:

1
./bin/utils/gen_snapshot --snapshot_kind=app-aot-elf --elf=libapp.so kernel.dill

经历重重关卡, 我们成功将dart源码转成了AOT快照, 那么接下来, 我们只需要使用gdb一步步跟踪快照的生成.

软路由折腾记录三

ADGuard

在路由器层面进行广告过滤, 并没有特别有效的方法, 到目前为止最有效的依旧是浏览器插件. 因为广告过滤的核心是拦截js/图片, 路由器上的透明代理很难窥视HTTPS传输的内容, 所以目前的路由器插件仅能对HTTP进行过滤. ADGuard Home另辟蹊径, 它提供DNS服务, 对黑名单里的广告域名进行拦截, 返回一个无法访问的IP.

部署

由于clash提供了DNS服务, 所以需要调整ADGuard/clash的上下游顺序.

1
dnsmasq -> ADGuard -> clash

首先启动ADGuard:

1
docker run --name adguardhome -v /my/own/workdir:/opt/adguardhome/work -v /my/own/confdir:/opt/adguardhome/conf -p 5301:53/tcp -p 5301:53/udp -p 3000:3000/tcp -d adguard/adguardhome

ADGuard的DNS服务监听5031端口, 而clash监听5300端口. 首先编辑dnsmasq配置文件, 设置上游服务器为ADGuard:

1
server=127.0.0.1#5301

接着访问软路由3000端口, 这是ADGuard提供的web界面, 在里面配置上游DNS为5300. 接下来就可以自有配置过滤规则, 配置完后可以执行命令验证广告域名是否被屏蔽:

1
dig ad.xxx.xxx

效果

由于广告推送技术的发展, 现在光屏蔽域名还远远不够. 实测效果来看, 国内各大视频网站的广告均无法过滤, 而类似于谷歌联盟的图片广告可以屏蔽. 另外ADGuard不仅仅用于广告过滤, 还能屏蔽一些统计js, 针对斗鱼/虎牙等还能屏蔽p2p通信, 这些都需要在web界面里勾选一些开源规则.
但需要注意, 屏蔽某些域名可能导致网站体验不佳, 例如禁用p2p后斗鱼/虎牙会偶尔卡顿.

ipset

在使用了tproxy透明代理UDP流量后, 所有TCP/UDP流量均由clash处理, 包括与大陆ip的通信. 由于UDP转发本身比较耗费CPU, 所以我考虑能不能让大陆ip直接在iptables层放行, 只有国外ip的流量会通过clash.
ipset是iptables提供的工具, 你可以给一组ip取一个名称, 然后在iptables规则中检测目标ip是否属于该组. 所以我们只需要把大陆的ip网段添加到一个组中, 取名为”china”, 接着新建规则, 如果ip属于”china”则不进行透明代理.

1
2
3
4
5
6
wget http://www.ipdeny.com/ipblocks/data/countries/cn.zone

ipset -N china hash:net
for i in $(cat ./cn.zone ); do
    ipset -A china $i;
done

接着在iptables相应的表中添加规则:

1
2
iptables -t mangle -A CLASH -p udp -m set --match-set china dst -j RETURN
iptables -t nat -A CLASH -p tcp -m set --match-set china dst -j RETURN

配置完成后, 只有国外ip的访问会经过clash, 而大陆ip的访问由iptables直接转发.

软路由折腾记录二

前言

我从大二开始接触linux, 最开始安装的就是基于debian的kali, 之后直接将debian作为主系统日常使用. 包括现在购买的VPS, 又或是wsl都安装的debian, 就如同我对firefox一样, 产生了深厚的感情.

改造

接上篇文章, 历经千幸万苦终于可以在OpenWrt上运行Docker后, tproxy透明代理失效了. 因为OpenWrt系统十分精简, 再加上是我自己编译的版本, 缺少官方的软件源, 所以调试变得异常困难. 最终我不得不放弃, 转向debian. 我熟练地下载官方镜像, 使用rufus写入U盘, 按部就班的完成了系统安装.
那么接下来就需要将它改造成路由器.

网桥

我的主机有4个网口, 我将第一个网口当做WAN, 连接着联通光猫. 那么我需要将剩下三个网口进行桥接, 使它们处于一个网段, 当为LAN口使用.

1
2
3
4
5
6
# 安装
apt install bridge-utils

# 添加桥接网卡
brctl addbr br0
brctl addif br0 enp0s8 enp0s9

你可以选择使用brctl手动添加桥接网卡, 但是重启之后会消失, 所以建议写入网络配置:

1
2
3
4
5
6
7
# /etc/network/interfaces
auto br0
iface br0 inet static
    address 192.168.12.1
    broadcast 192.168.12.255
    netmask 255.255.255.0
    bridge-ports enp0s8 enp0s9

bridge-ports参数替换成具体的网卡, 上面的例子只桥接了两个网卡, 这并不是我实际使用的配置, 仅供参考.

DNS/DHCP

路由器最重要的自然是DHCP/DNS, 而dnsmasq帮我们轻松搞定:

1
apt install dnsmasq

dnsmasq会提供DHCP/DNS服务, DNS默认转发到上游DNS, 暂时不需要修改, 但是DHCP需要修改配置文件指定网段等信息:

1
2
3
# /etc/dnsmasq.conf
interface=br0
dhcp-range=192.168.12.100,192.168.12.250,72h

转发

现在将电脑网线插入J1900主机网口, 电脑可以使用DHCP获取到IP, 可惜现在还上不了网. 因为linux默认将目标IP不是本机的流量全部DROP, 首先打开IP转发:

1
2
# 重启后失效
echo 1 > /proc/sys/net/ipv4/ip_forward

光开启转发还是无法上网, 因为内网网段访问外网需要NAT, 所以需要配置iptables:

1
iptables -t nat -A POSTROUTING -s 192.168.12.0/24 -o enp0s3 -j MASQUERADE

enp0s3是WAN口网卡, MASQUERADE将把子网流量自动NAT转换, 现在你可以自由的进行网上冲浪了.

复现

完成上面一切, debian现在与OpenWrt有着同样的功能, 而且还废弃了那些累赘的组件. 我现在准备安装docker-ce, 过程很迅速, 安装完后我开始基于docker-ce搭建自己的SMB服务.
一切都完成了, 我长舒一口气, 我在终端执行

1
dig google.com @8.8.8.8

命令卡住了!!! tproxy又罢工了!!! 我突然意识到我错怪了OpenWrt, 这似乎是Docker惹的祸, 于是我又开始了漫长的debug.

调试

网络架构


这张图显示的是linux的网络架构, 包括iptables的链条顺序、路由选择, 下面所有的调试都是基于这张图.

Docker

我知道Docker的网络建立在iptables上, 它一样使用虚拟桥接网卡连接各个容器, 然后在iptables添加规则进行NAT. 可是没理由它会影响到主机的网络, 我一条一条地查看Docker添加的防火墙规则, 唯一异常的点是它将filter表的FORWARD默认策略改成了DROP. 但tproxy压根不会经过FORWARD, 路由规则会让流量进入本地协议栈.
根据架构图可以知道, 如果包进入了filter表的INPUT链, 那么说明流量已经过了路由选择, 即将进入本地协议栈. 如果tproxy正常工作, 修改了包的结构体, 那么此时代理端口与包是相关联的, 所以包会被正常接收.
我开始猜测tproxy没有正常运作, 导致路由失败, 包被FORWARD出去了:

1
iptables -t filter -I INPUT 1 -p udp -j LOG --log-prefix '** SUSPECT filter INPUT**'

我添加了一条iptables规则, 当匹配到包时会打印出日志, 于是我重复了一次DNS查询操作. 日志出现了, 所以我断定它已经走过了路由选择. 我又接着猜测了几种可能:

  • tproxy关联包信息失败
  • IP_TRANSPARENT属性失效

这两种情况我都没有有效的测试手段, 所以我陷入了困境, 不过我觉得原因大概率是docker-ce加载了某个内核模块. 于是我使用dpkg查看所有安装的包, 只发现了一个aufs.ko, 貌似不是很符合.
当我束手无策时, 我发现了这篇帖子TPROXY compatibility with Docker.
作者遇到了和我一样的问题, 于是它使用strace跟踪dockerd的系统调用, 发现它加载了一个内核模块”br_netfilter”. 我不抱任何希望的执行了”rmmod br_netfilter”, 它恢复了!!! 原来是这个br_netfilter的问题!!!

ebtables

这时我才仔细去观察上面的架构图, 我发现我遗漏了一个问题, 就是网桥的流量会走Link Layer, 所以可能是这部分出了问题. 图中的蓝色表代表的是ebtables, 绿色表代表iptables, ebtables可以理解成链路的iptables.
我想弄清楚流量的具体走向, 所以我在每个点都加上log规则, 数据包的流向将一览无遗:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# raw
iptables -t raw -I PREROUTING 1 -p udp -j LOG --log-prefix '** SUSPECT raw PREROUTING**'
iptables -t raw -I OUTPUT 1 -p udp -j LOG --log-prefix '** SUSPECT raw OUTPUT**'

# mangle
iptables -t mangle -I PREROUTING 1 -p udp -j LOG --log-prefix '** SUSPECT mangle PREROUTING**'
iptables -t mangle -I INPUT 1 -p udp -j LOG --log-prefix '** SUSPECT mangle INPUT**'
iptables -t mangle -I FORWARD 1 -p udp -j LOG --log-prefix '** SUSPECT mangle FORWARD**'

# nat
iptables -t nat -I PREROUTING 1 -p udp -j LOG --log-prefix '** SUSPECT nat PREROUTING**'
iptables -t nat -I INPUT 1 -p udp -j LOG --log-prefix '** SUSPECT nat INPUT**'

# filter
iptables -t filter -I INPUT 1 -p udp -j LOG --log-prefix '** SUSPECT filter INPUT**'
iptables -t filter -I FORWARD 1 -p udp -j LOG --log-prefix '** SUSPECT filter FORWARD**'

# BROUTING
ebtables-legacy -t broute -A BROUTING -p ipv4 --ip-proto udp --log-level 6 --log-ip --log-prefix "TRACE: eb:broute:BROUTING" -j CONTINUE

# nat
ebtables-legacy -t nat -A OUTPUT -p ipv4 --ip-proto udp --log-level 6 --log-ip --log-prefix "TRACE: eb:nat:OUTPUT"  -j CONTINUE
ebtables-legacy -t nat -A PREROUTING -p ipv4 --ip-proto udp --log-level 6 --log-ip --log-prefix "TRACE: eb:nat:PREROUTING" -j CONTINUE
ebtables-legacy -t nat -A POSTROUTING -p ipv4 --ip-proto udp --log-level 6 --log-ip --log-prefix "TRACE: eb:nat:POSTROUTING" -j CONTINUE

# filter
ebtables-legacy -t filter -A INPUT -p ipv4 --ip-proto udp --log-level 6 --log-ip --log-prefix "TRACE: eb:filter:INPUT" -j CONTINUE
ebtables-legacy -t filter -A FORWARD -p ipv4 --ip-proto udp --log-level 6 --log-ip --log-prefix "TRACE: eb:filter:FORWARD" -j CONTINUE
ebtables-legacy -t filter -A OUTPUT -p ipv4 --ip-proto udp --log-level 6 --log-ip --log-prefix "TRACE: eb:filter:OUTPUT" -j CONTINUE

在未加载br_netfilter的情况下, 我向8.8.8.8查询域名:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
TRACE: eb:broute:BROUTING IN=enp0s8 OUT= MAC source = 08:00:27:6d:c8:0b MAC dest = 08:00:27:9b:fa:c9 proto = 0x0800 IP SRC=192.168.12.192 IP DST=8.8.8.8, IP tos=0x00, IP proto=17 SPT=36289 DPT=53
TRACE: eb:nat:PREROUTING IN=enp0s8 OUT= MAC source = 08:00:27:6d:c8:0b MAC dest = 08:00:27:9b:fa:c9 proto = 0x0800 IP SRC=192.168.12.192 IP DST=8.8.8.8, IP tos=0x00, IP proto=17 SPT=36289 DPT=53
TRACE: eb:filter:INPUT IN=enp0s8 OUT= MAC source = 08:00:27:6d:c8:0b MAC dest = 08:00:27:9b:fa:c9 proto = 0x0800 IP SRC=192.168.12.192 IP DST=8.8.8.8, IP tos=0x00, IP proto=17 SPT=36289 DPT=53
** SUSPECT raw PREROUTING**IN=br0 OUT= MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=24626 PROTO=UDP SPT=36289 DPT=53 LEN=46
** SUSPECT mangle PREROUTING*IN=br0 OUT= MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=24626 PROTO=UDP SPT=36289 DPT=53 LEN=46
** SUSPECT nat PREROUTING**IN=br0 OUT= MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=24626 PROTO=UDP SPT=36289 DPT=53 LEN=46 MARK=0x162
** SUSPECT mangle INPUT**IN=br0 OUT= MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=24626 PROTO=UDP SPT=36289 DPT=53 LEN=46 MARK=0x162
** SUSPECT filter INPUT**IN=br0 OUT= MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=24626 PROTO=UDP SPT=36289 DPT=53 LEN=46 MARK=0x162
** SUSPECT nat INPUT**IN=br0 OUT= MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=24626 PROTO=UDP SPT=36289 DPT=53 LEN=46 MARK=0x162
** SUSPECT raw OUTPUT**IN= OUT=enp0s3 SRC=10.0.2.15 DST=10.91.0.1 LEN=57 TOS=0x00 PREC=0x00 TTL=64 ID=56764 DF PROTO=UDP SPT=45711 DPT=53 LEN=37
** SUSPECT raw OUTPUT**IN= OUT=enp0s3 SRC=10.0.2.15 DST=10.91.0.1 LEN=57 TOS=0x00 PREC=0x00 TTL=64 ID=56765 DF PROTO=UDP SPT=59060 DPT=53 LEN=37
** SUSPECT raw PREROUTING**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=114 TOS=0x00 PREC=0x00 TTL=64 ID=1722 PROTO=UDP SPT=53 DPT=45711 LEN=94
** SUSPECT mangle PREROUTING*IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=114 TOS=0x00 PREC=0x00 TTL=64 ID=1722 PROTO=UDP SPT=53 DPT=45711 LEN=94
** SUSPECT mangle INPUT**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=114 TOS=0x00 PREC=0x00 TTL=64 ID=1722 PROTO=UDP SPT=53 DPT=45711 LEN=94
** SUSPECT filter INPUT**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=114 TOS=0x00 PREC=0x00 TTL=64 ID=1722 PROTO=UDP SPT=53 DPT=45711 LEN=94
** SUSPECT raw PREROUTING**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=73 TOS=0x00 PREC=0x00 TTL=64 ID=1723 PROTO=UDP SPT=53 DPT=59060 LEN=53
** SUSPECT mangle PREROUTING*IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=73 TOS=0x00 PREC=0x00 TTL=64 ID=1723 PROTO=UDP SPT=53 DPT=59060 LEN=53
** SUSPECT mangle INPUT**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=73 TOS=0x00 PREC=0x00 TTL=64 ID=1723 PROTO=UDP SPT=53 DPT=59060 LEN=53
** SUSPECT filter INPUT**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=73 TOS=0x00 PREC=0x00 TTL=64 ID=1723 PROTO=UDP SPT=53 DPT=59060 LEN=53
** SUSPECT raw OUTPUT**IN= OUT=br0 SRC=8.8.8.8 DST=192.168.12.192 LEN=98 TOS=0x00 PREC=0x00 TTL=64 ID=40027 DF PROTO=UDP SPT=53 DPT=36289 LEN=78
TRACE: eb:nat:OUTPUT IN= OUT=enp0s8 MAC source = 08:00:27:9b:fa:c9 MAC dest = 08:00:27:6d:c8:0b proto = 0x0800 IP SRC=8.8.8.8 IP DST=192.168.12.192, IP tos=0x00, IP proto=17 SPT=53 DPT=36289
TRACE: eb:filter:OUTPUT IN= OUT=enp0s8 MAC source = 08:00:27:9b:fa:c9 MAC dest = 08:00:27:6d:c8:0b proto = 0x0800 IP SRC=8.8.8.8 IP DST=192.168.12.192, IP tos=0x00, IP proto=17 SPT=53 DPT=36289
TRACE: eb:nat:POSTROUTING IN= OUT=enp0s8 MAC source = 08:00:27:9b:fa:c9 MAC dest = 08:00:27:6d:c8:0b proto = 0x0800 IP SRC=8.8.8.8 IP DST=192.168.12.192, IP tos=0x00, IP proto=17 SPT=53 DPT=36289

可以看到包先过了ebtables规则, 然后转入了iptables规则, 也就是包按层级从下到上. 此时的tproxy是正常工作的, 可能你会感到疑惑, 为什么本机向10.91.0.1发送了UDP. 其实是因为我使用了VPN, 所以这是在解析VPN服务器的域名, UDP的转发走的是VPN的TCP通道.
接着我加载了br_netfilter, 观察日志:

1
2
3
4
5
6
7
8
9
TRACE: eb:broute:BROUTING IN=enp0s8 OUT= MAC source = 08:00:27:6d:c8:0b MAC dest = 08:00:27:9b:fa:c9 proto = 0x0800 IP SRC=192.168.12.192 IP DST=8.8.8.8, IP tos=0x00, IP proto=17 SPT=36003 DPT=53
TRACE: eb:nat:PREROUTING IN=enp0s8 OUT= MAC source = 08:00:27:6d:c8:0b MAC dest = 08:00:27:9b:fa:c9 proto = 0x0800 IP SRC=192.168.12.192 IP DST=8.8.8.8, IP tos=0x00, IP proto=17 SPT=36003 DPT=53
** SUSPECT raw PREROUTING**IN=br0 OUT= PHYSIN=enp0s8 MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=35740 PROTO=UDP SPT=36003 DPT=53 LEN=46
** SUSPECT mangle PREROUTING*IN=br0 OUT= PHYSIN=enp0s8 MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=35740 PROTO=UDP SPT=36003 DPT=53 LEN=46
** SUSPECT nat PREROUTING**IN=br0 OUT= PHYSIN=enp0s8 MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=35740 PROTO=UDP SPT=36003 DPT=53 LEN=46 MARK=0x162
TRACE: eb:filter:INPUT IN=enp0s8 OUT= MAC source = 08:00:27:6d:c8:0b MAC dest = 08:00:27:9b:fa:c9 proto = 0x0800 IP SRC=192.168.12.192 IP DST=8.8.8.8, IP tos=0x00, IP proto=17 SPT=36003 DPT=53
** SUSPECT mangle INPUT**IN=br0 OUT= PHYSIN=enp0s8 MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=35740 PROTO=UDP SPT=36003 DPT=53 LEN=46 MARK=0x162
** SUSPECT filter INPUT**IN=br0 OUT= PHYSIN=enp0s8 MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=35740 PROTO=UDP SPT=36003 DPT=53 LEN=46 MARK=0x162
** SUSPECT nat INPUT**IN=br0 OUT= PHYSIN=enp0s8 MAC=08:00:27:9b:fa:c9:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=35740 PROTO=UDP SPT=36003 DPT=53 LEN=46 MARK=0x162

这是什么情况??? 数据包在两个层级之间跳来跳去??? 我开始搜索相关资料, 原来这是br_netfilter提供的透明防火墙功能, 使得三层规则可以作用在二层上. 因为桥接的端口之间发送包是被网桥直接转走的, 不会走到本机的iptables中, 而br_netfilter使得ebtables主动调用iptables规则, 这样所有的包都能接受过滤.
谜底揭开了, 我看到的日志显示包已经到了filter表的INPUT, 可实际上它还没有过IP层路由选择, 它此时在二层ebtables的filter表中, 只不过它主动的调用了iptables. 也就是说这个包没有通过路由进入本地协议栈, 它被丢弃或者转走了. 大概是因为iptables打上的fwmark, 在ebtables中丢失了, 导致进行路由选择时无法命中我们添加的路由规则.
于是我看阅读ebtables的文档, 文档中说道, 在broute表中将流量DROP, 会使得包进入Network Layer处理, 这样就是图中broute有个箭头指到上层的含义.

1
ebtables-legacy -t broute -A BROUTING -i enp0s8 -p ipv4 --ip-proto udp -j redirect --redirect-target DROP

我执行上面的命令后, 再执行查询操作:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
TRACE: eb:broute:BROUTING IN=enp0s8 OUT= MAC source = 08:00:27:6d:c8:0b MAC dest = 08:00:27:9b:fa:c9 proto = 0x0800 IP SRC=192.168.12.192 IP DST=8.8.8.8, IP tos=0x00, IP proto=17 SPT=47382 DPT=53
** SUSPECT raw PREROUTING**IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=25647 PROTO=UDP SPT=47382 DPT=53 LEN=46
** SUSPECT mangle PREROUTING*IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=25647 PROTO=UDP SPT=47382 DPT=53 LEN=46
** SUSPECT nat PREROUTING**IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=25647 PROTO=UDP SPT=47382 DPT=53 LEN=46 MARK=0x162
** SUSPECT mangle INPUT**IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=25647 PROTO=UDP SPT=47382 DPT=53 LEN=46 MARK=0x162
** SUSPECT filter INPUT**IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=25647 PROTO=UDP SPT=47382 DPT=53 LEN=46 MARK=0x162
** SUSPECT nat INPUT**IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=8.8.8.8 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=25647 PROTO=UDP SPT=47382 DPT=53 LEN=46 MARK=0x162
** SUSPECT raw OUTPUT**IN= OUT=enp0s3 SRC=10.0.2.15 DST=10.91.0.1 LEN=57 TOS=0x00 PREC=0x00 TTL=64 ID=40282 DF PROTO=UDP SPT=54985 DPT=53 LEN=37
** SUSPECT raw OUTPUT**IN= OUT=enp0s3 SRC=10.0.2.15 DST=10.91.0.1 LEN=57 TOS=0x00 PREC=0x00 TTL=64 ID=40283 DF PROTO=UDP SPT=37997 DPT=53 LEN=37
** SUSPECT raw PREROUTING**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=73 TOS=0x00 PREC=0x00 TTL=64 ID=4012 PROTO=UDP SPT=53 DPT=37997 LEN=53
** SUSPECT mangle PREROUTING*IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=73 TOS=0x00 PREC=0x00 TTL=64 ID=4012 PROTO=UDP SPT=53 DPT=37997 LEN=53
** SUSPECT mangle INPUT**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=73 TOS=0x00 PREC=0x00 TTL=64 ID=4012 PROTO=UDP SPT=53 DPT=37997 LEN=53
** SUSPECT filter INPUT**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=73 TOS=0x00 PREC=0x00 TTL=64 ID=4012 PROTO=UDP SPT=53 DPT=37997 LEN=53
** SUSPECT raw PREROUTING**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=114 TOS=0x00 PREC=0x00 TTL=64 ID=4018 PROTO=UDP SPT=53 DPT=54985 LEN=94
** SUSPECT mangle PREROUTING*IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=114 TOS=0x00 PREC=0x00 TTL=64 ID=4018 PROTO=UDP SPT=53 DPT=54985 LEN=94
** SUSPECT mangle INPUT**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=114 TOS=0x00 PREC=0x00 TTL=64 ID=4018 PROTO=UDP SPT=53 DPT=54985 LEN=94
** SUSPECT filter INPUT**IN=enp0s3 OUT= MAC=08:00:27:4f:1e:2b:52:54:00:12:35:02:08:00 SRC=10.91.0.1 DST=10.0.2.15 LEN=114 TOS=0x00 PREC=0x00 TTL=64 ID=4018 PROTO=UDP SPT=53 DPT=54985 LEN=94
** SUSPECT raw OUTPUT**IN= OUT=br0 SRC=8.8.8.8 DST=192.168.12.192 LEN=98 TOS=0x00 PREC=0x00 TTL=64 ID=26403 DF PROTO=UDP SPT=53 DPT=47382 LEN=78
TRACE: eb:nat:OUTPUT IN= OUT=enp0s8 MAC source = 08:00:27:9b:fa:c9 MAC dest = 08:00:27:6d:c8:0b proto = 0x0800 IP SRC=8.8.8.8 IP DST=192.168.12.192, IP tos=0x00, IP proto=17 SPT=53 DPT=47382
TRACE: eb:filter:OUTPUT IN= OUT=enp0s8 MAC source = 08:00:27:9b:fa:c9 MAC dest = 08:00:27:6d:c8:0b proto = 0x0800 IP SRC=8.8.8.8 IP DST=192.168.12.192, IP tos=0x00, IP proto=17 SPT=53 DPT=47382
TRACE: eb:nat:POSTROUTING IN= OUT=enp0s8 MAC source = 08:00:27:9b:fa:c9 MAC dest = 08:00:27:6d:c8:0b proto = 0x0800 IP SRC=8.8.8.8 IP DST=192.168.12.192, IP tos=0x00, IP proto=17 SPT=53 DPT=47382

可以看到包的确被重定向到了Network Layer, tproxy也正常工作了.

尾声

正当我以为一切都结束了的时候, 我打开了斗鱼直播, 可浏览器却显示网络异常, 我又上不了网了. 我在终端进行ping测试, IP可以ping通而域名不行, 这样看来是DNS的问题.

1
dig baidu.com @192.168.12.1

命令卡住了, 难道发往本地的包重定向到Network Layer就被拒收了? 我又开始观察ebtables/iptables的日志输出:

1
2
3
4
5
6
7
TRACE: eb:broute:BROUTING IN=enp0s8 OUT= MAC source = 08:00:27:6d:c8:0b MAC dest = 08:00:27:9b:fa:c9 proto = 0x0800 IP SRC=192.168.12.192 IP DST=192.168.12.1, IP tos=0x00, IP proto=17 SPT=34741 DPT=53
** SUSPECT raw PREROUTING**IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=192.168.12.1 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=7347 PROTO=UDP SPT=34741 DPT=53 LEN=46
** SUSPECT mangle PREROUTING*IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=192.168.12.1 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=7347 PROTO=UDP SPT=34741 DPT=53 LEN=46
** SUSPECT nat PREROUTING**IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=192.168.12.1 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=7347 PROTO=UDP SPT=34741 DPT=53 LEN=46
** SUSPECT mangle INPUT**IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=192.168.12.1 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=7347 PROTO=UDP SPT=34741 DPT=53 LEN=46
** SUSPECT filter INPUT**IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=192.168.12.1 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=7347 PROTO=UDP SPT=34741 DPT=53 LEN=46
** SUSPECT nat INPUT**IN=enp0s8 OUT= MAC=08:00:27:ae:7d:59:08:00:27:6d:c8:0b:08:00 SRC=192.168.12.192 DST=192.168.12.1 LEN=66 TOS=0x00 PREC=0x00 TTL=64 ID=7347 PROTO=UDP SPT=34741 DPT=53 LEN=46

数据包已经进入了INPUT, 这次的确是进入了本地协议栈, 但是需要注意数据包的IN网卡已经从桥接网卡bro变成了真实网卡端口enp0s8. 难道是因为这个改变, 而被协议栈拒收? 还是因为dnsmasq没有bind该网卡?
我手动使用nc进行测试:

1
nc -u -lp 3000

使用nc监听软路由上3000端口, 然后在电脑上执行:

1
dig baidu.com @192.168.12.1 -p 3000

如果数据能被接收, nc会打印出DNS查询报文, 结果令我诧异, nc打印出了报文内容. 那么看来即使IN interface被修改, 数据包依旧能够被正确接收.
最后我使用strace追踪dnsmasq的系统调用:

1
strace /sbin/dnsmasq

执行查询后, 我发现dnsmasq接收到了报文, 只是它似乎获取了网卡名称bro与enp0s8, 应该是它进行了网卡校验, 认为请求不应该由enp0s8流入, 所以没有进行回应.

总结

为了不影响Docker的正常使用, 我没有将所有Link Layer流量进行重定向, 放行了一些常见内网IP网段.
所以我最后的ebtables/iptables配置如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
#!/bin/sh
# tcp
iptables -t nat -N CLASH
iptables -t nat -A CLASH -d 127.0.0.0/8 -j RETURN
iptables -t nat -A CLASH -d 10.0.0.0/8 -j RETURN
iptables -t nat -A CLASH -d 169.254.0.0/16 -j RETURN
iptables -t nat -A CLASH -d 192.168.0.0/16 -j RETURN
iptables -t nat -A CLASH -d 224.0.0.0/4 -j RETURN
iptables -t nat -A CLASH -d 240.0.0.0/4 -j RETURN
iptables -t nat -A CLASH -d 172.16.0.0/12 -j RETURN
iptables -t nat -A CLASH -d 255.255.255.255 -j RETURN
iptables -t nat -A CLASH -p tcp -j REDIRECT --to-ports 7892
iptables -t nat -A PREROUTING -p tcp -j CLASH

# udp
ip rule add fwmark 0x162 table 0x162
ip route add local 0.0.0.0/0 dev lo table 0x162
iptables -t mangle -N CLASH
iptables -t mangle -A CLASH -d 127.0.0.0/8 -j RETURN
iptables -t mangle -A CLASH -d 10.0.0.0/8 -j RETURN
iptables -t mangle -A CLASH -d 169.254.0.0/16 -j RETURN
iptables -t mangle -A CLASH -d 192.168.0.0/16 -j RETURN
iptables -t mangle -A CLASH -d 224.0.0.0/4 -j RETURN
iptables -t mangle -A CLASH -d 240.0.0.0/4 -j RETURN
iptables -t mangle -A CLASH -d 172.16.0.0/12 -j RETURN
iptables -t mangle -A CLASH -d 255.255.255.255 -j RETURN
iptables -t mangle -A CLASH -p udp -j TPROXY --on-port 7892 --tproxy-mark 0x162
iptables -t mangle -A PREROUTING -p udp -j CLASH

ebtables-legacy -t broute -A BROUTING -p ipv4 --ip-proto udp --ip-destination 127.0.0.0/8 -j ACCEPT
ebtables-legacy -t broute -A BROUTING -p ipv4 --ip-proto udp --ip-destination 10.0.0.0/8 -j ACCEPT
ebtables-legacy -t broute -A BROUTING -p ipv4 --ip-proto udp --ip-destination 169.254.0.0/16 -j ACCEPT
ebtables-legacy -t broute -A BROUTING -p ipv4 --ip-proto udp --ip-destination 192.168.0.0/16 -j ACCEPT
ebtables-legacy -t broute -A BROUTING -p ipv4 --ip-proto udp --ip-destination 224.0.0.0/4 -j ACCEPT
ebtables-legacy -t broute -A BROUTING -p ipv4 --ip-proto udp --ip-destination 240.0.0.0/4 -j ACCEPT
ebtables-legacy -t broute -A BROUTING -p ipv4 --ip-proto udp --ip-destination 172.16.0.0/12 -j ACCEPT
ebtables-legacy -t broute -A BROUTING -p ipv4 --ip-proto udp --ip-destination 255.255.255.255 -j ACCEPT
ebtables-legacy -t broute -A BROUTING -p ipv4 --ip-proto udp -j redirect --redirect-target DROP

有一点需要注意, ebtables规则不需要”-i name”指定网卡名称, 因为理论上docker容器与个人电脑(连接软路由LAN口)处于同一网络层面. 不管是LAN口网桥还是docker网桥, 都需要适配该规则, 否则UDP透明代理将失败.