From e8a579f7c90ab32d10078ef51d9b12da5895c433 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 2 Aug 2026 22:55:17 +0000 Subject: [PATCH] grpc interceptors, and channel-wide credentials MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit net.grpc had retryable_code and nothing else, while std.web had a middleware chain with rate_limit and circuit ready made — so there was nowhere to put auth on a grpc transport short of writing the check into every method. an interceptor now wraps a dispatch the way a web middleware wraps a route handler: it takes the rest of the chain plus the call, so it can run code before or after, or answer the call and never reach the dispatch. intercept() folds a list of them into a dispatch, which is what serve() already takes, so no serve form changes signature and a generated serve_ router composes exactly like a hand-written one. the Chain struct is the same shape web's uses, and for the same reason: a closure capturing a bare fn-value miscompiles, one capturing a struct it calls in place does not. authorize() is the hook the gap was really about — it runs against the caller's metadata before any method body, so one interceptor covers every rpc a server exposes, and bearer_token() reads the credential out of an authorization header with the case-insensitive scheme match rfc 7235 asks for. rate_limit() and circuit() take the same std.resilience Limiter and Breaker the web middlewares take, so one limiter can cap an http and a grpc surface together. circuit() counts only the server's own faults: a client sending bad requests is not a reason to stop serving good ones. on the client, set_credentials() attaches metadata to a channel instead of to every call site. it mutates rather than returning a configured copy — two Conn values sharing one http/2 client would each close it — and per-call metadata is appended after it, so a call can add to the credentials but not silently drop them. an interceptor refuses a call by returning refuse(status, message) rather than failing: a pith lambda can neither fail nor use !, which is why the ready-made ones are shaped the way they are. verified: 8 unit tests over composition order, refusal, error propagation, and each ready-made interceptor; a round-trip case where a real client is refused UNAUTHENTICATED without a credential and served with one; and a runnable example putting auth, a cap, and a counter over a service that knows about none of them. --- docs/grpc.md | 69 +++++- examples/expected/grpc_interceptors.txt | 5 + examples/grpc_interceptors.pith | 97 ++++++++ std/net/grpc.pith | 277 +++++++++++++++++++++- tests/cases/test_grpc_interceptors.pith | 84 +++++++ tests/expected/test_grpc_interceptors.txt | 2 + 6 files changed, 528 insertions(+), 6 deletions(-) create mode 100644 examples/expected/grpc_interceptors.txt create mode 100644 examples/grpc_interceptors.pith create mode 100644 tests/cases/test_grpc_interceptors.pith create mode 100644 tests/expected/test_grpc_interceptors.txt diff --git a/docs/grpc.md b/docs/grpc.md index f54405fb..464f5836 100644 --- a/docs/grpc.md +++ b/docs/grpc.md @@ -308,6 +308,68 @@ what arrived, or bytes that were never framed at all — fails `INVALID_ARGUMENT` rather than decoding as an all-default message. a unary request carrying more than one message fails the same way. +## interceptors + +an interceptor wraps the dispatch the way a `std.web` middleware wraps a route +handler. it takes the rest of the chain plus the call, so it can run code +before or after, or answer the call itself and never reach the dispatch: + +```pith +fn log_calls(next: fn(String, Bytes) -> Bytes!grpc.GrpcError, path: String, request: Bytes) -> Bytes!grpc.GrpcError: + print(path) + return next(path, request) + +guarded := grpc.intercept(serve_Chat, [grpc.authorize(check_token), log_calls]) +grpc.serve("0.0.0.0", 50051, guarded)! +``` + +`intercept` returns a dispatch, so it drops into `serve`, `serve_tls`, or any +other serve form without changing a signature — a generated `serve_` router +and a hand-written dispatch both compose the same way. the first interceptor in +the list runs outermost, seeing the call first and the reply last. + +three come ready made: + +- `grpc.authorize(verify)` runs `verify` against the caller's metadata before + any method body, and answers `UNAUTHENTICATED` when it returns false. this is + the transport-level auth hook: one interceptor covers every rpc the server + exposes. `grpc.bearer_token(metadata)` pulls the token out of an + `authorization` header (matching the scheme case-insensitively, as rfc 7235 + requires), so verifying a jwt is a one-liner against `std.crypto.jwt`. +- `grpc.rate_limit(limiter)` spends a token per call and answers + `RESOURCE_EXHAUSTED` when the bucket is dry. +- `grpc.circuit(breaker)` opens on repeated *server* faults — `INTERNAL`, + `UNAVAILABLE`, `DATA_LOSS`, an expired deadline — and answers `UNAVAILABLE` + while open. a caller's own error (`NOT_FOUND`, an invalid argument, a refused + credential) does not count against it: a client sending bad requests is not a + reason to stop serving good ones. + +the last two take the same `std.resilience` `Limiter` and `Breaker` that +`web.rate_limit` and `web.circuit` take, so one limiter can cap an http surface +and a grpc surface together. + +`examples/grpc_interceptors.pith` runs the whole shape end to end — a service +that knows nothing about auth or rate limiting, guarded by both, called by a +client that presents its credential once. + +writing your own: an interceptor that refuses a call returns +`grpc.refuse(status, message)` rather than failing directly, because a pith +lambda can neither `fail` nor use `!`. return `next(path, request)` to pass the +call along. + +on the client, a credential belongs on the channel rather than on every call +site: + +```pith +conn := grpc.dial_h2c("localhost", 50051)! +conn.set_credentials(["authorization", "Bearer " + token]) +reply := conn.unary("/chat.Chat/Send", req)! # carries the credential +``` + +per-call metadata still works through the `*_with_headers` forms and is appended +after the channel's, so a call can add to the credentials but not silently drop +them. + ## what isn't here yet - **generated streaming is server-side.** the generated *client* still collects @@ -322,4 +384,9 @@ request carrying more than one message fails the same way. are enforced at the edges (client timeout, server expiry-on-arrival), not by cancelling a handler mid-flight. - **observability**: the client opens a trace span and records red metrics per - call automatically; the server side is your code, so instrument it yourself. + call automatically. the server side is your code — an interceptor is the + place to put it, since it sees every method. +- **interceptors are server-side and unary.** the client has channel-wide + credentials but no interceptor chain of its own, and a server interceptor + wraps the unary dispatch; the streaming serve forms take their own dispatch + shapes and are not wrapped by `intercept`. diff --git a/examples/expected/grpc_interceptors.txt b/examples/expected/grpc_interceptors.txt new file mode 100644 index 00000000..cc96c3de --- /dev/null +++ b/examples/expected/grpc_interceptors.txt @@ -0,0 +1,5 @@ +anonymous: UNAUTHENTICATED +with token: hello, ada +with token: hello, grace +over the cap: RESOURCE_EXHAUSTED +calls seen: 4 diff --git a/examples/grpc_interceptors.pith b/examples/grpc_interceptors.pith new file mode 100644 index 00000000..4b1c45f7 --- /dev/null +++ b/examples/grpc_interceptors.pith @@ -0,0 +1,97 @@ +# grpc interceptors: cross-cutting rules that run before any method body. +# +# an interceptor wraps the dispatch the way a std.web middleware wraps a route +# handler — it takes the rest of the chain plus the call, so it can inspect the +# call, wrap the reply, or refuse outright and never reach the dispatch. +# intercept() folds a list of them into a dispatch, which is what serve() takes, +# so nothing about the service signature changes. +# +# this server carries two rules a real service wants and neither method knows +# about: a bearer credential is required, and calls are capped. the client +# presents its credential once, on the channel, rather than at every call site. + +import std.net.grpc as grpc +import std.resilience as resilience +import std.protobuf as protobuf +import std.bytes as bytes +import std.time as time + +PORT := 50153 +TOKEN := "letmein" + +fn read_text(message: Bytes) -> String!protobuf.ProtoError: + r := protobuf.reader(message) + mut text := "" + while not r.at_end(): + tag := r.read_tag()! + if tag.field == 1: + text = r.read_string()! + else: + r.skip(tag.wire)! + return text + +fn message_of(text: String) -> Bytes: + w := protobuf.writer() + w.write_string(1, text) catch false + out := w.bytes() + w.free() + return out + +# the service itself: no auth code, no rate-limit code, no idea either exists. +fn greeter(path: String, request: Bytes) -> Bytes!grpc.GrpcError: + if path != "/demo.Greeter/Hello": + fail grpc.GrpcError(grpc.GRPC_UNIMPLEMENTED, "no method " + path) + name := read_text(request) catch "world" + return message_of("hello, " + name) + +# the auth rule, checked against the metadata the caller put on the wire. a real +# server would verify a jwt here — std.crypto.jwt.verify_hs256(token, secret) — +# rather than compare a fixed string. +fn check_token(metadata: Map[String, String]) -> Bool: + return grpc.bearer_token(metadata) == TOKEN + +# an interceptor of our own: count every call, whatever the outcome. +mut served := 0 + +fn count_calls(next: fn(String, Bytes) -> Bytes!grpc.GrpcError, path: String, request: Bytes) -> Bytes!grpc.GrpcError: + served = served + 1 + return next(path, request) + +fn run_server(): + # a reserve of two and a slow refill, so the third call in this run is + # refused rather than racing a token that trickles back in mid-run. + limiter := resilience.rate_limiter(1, 2) + # the first interceptor runs outermost, so calls are counted even when the + # credential check below refuses them. + guarded := grpc.intercept(greeter, [count_calls, grpc.authorize(check_token), grpc.rate_limit(limiter)]) + grpc.serve("127.0.0.1", PORT, guarded) catch 0 + +# call once and describe what came back. +fn say_hello(conn: grpc.Conn, name: String) -> String: + reply := conn.unary("/demo.Greeter/Hello", message_of(name)) + if reply.is_err: + return grpc.status_name(reply.err.status) + return read_text(reply.ok) catch "?" + +fn main() -> Int!: + spawn run_server() + time.delay(300) + + # no credential: the authorize interceptor refuses before greeter runs + anon := grpc.dial_h2c("127.0.0.1", PORT)! + print("anonymous: " + say_hello(anon, "ada")) + anon.close() + + # the credential rides the channel, so every call over it carries the token + conn := grpc.dial_h2c("127.0.0.1", PORT)! + conn.set_credentials(["authorization", "Bearer " + TOKEN]) + print("with token: " + say_hello(conn, "ada")) + print("with token: " + say_hello(conn, "grace")) + + # the burst is spent, so the limiter refuses the next one + print("over the cap: " + say_hello(conn, "alan")) + conn.close() + + # count_calls sat outside the other two, so it saw all four + print("calls seen: {served}") + return 0 diff --git a/std/net/grpc.pith b/std/net/grpc.pith index d2b1324c..2a882f11 100644 --- a/std/net/grpc.pith +++ b/std/net/grpc.pith @@ -129,8 +129,42 @@ pub fn status_name(status: Int) -> String: pub struct Conn: client: http2.Client authority: String + # metadata sent on every call over this channel — credentials, usually. + # set_credentials fills it; empty means the channel adds nothing. + credentials: List[String] + +# an empty credential list, typed for the Conn field. +fn no_credentials() -> List[String]: + out: List[String] := [] + return out impl Conn: + ## send `metadata` — flat key, value, key, value — on every call over this + ## channel, so a credential is attached once at dial instead of threaded + ## through every call site: + ## + ## conn := grpc.dial_h2c("localhost", 50051)! + ## conn.set_credentials(["authorization", "Bearer " + token]) + ## + ## this mutates the channel rather than returning a configured copy: two + ## Conn values sharing one http/2 client would each close it. per-call + ## metadata still goes through the *_with_headers forms, and is appended + ## after these, so a call can add to them but not silently drop them. + pub fn set_credentials(metadata: List[String]): + self.credentials = metadata + + # this channel's credentials followed by `extra`, which is what every call + # actually sends. + fn call_metadata(extra: List[String]) -> List[String]: + if self.credentials.len() == 0: + return extra + mut out: List[String] := [] + for item in self.credentials: + out.push(item) + for item in extra: + out.push(item) + return out + # run a unary rpc over this channel. `full_method` is "/package.Service/Method" # and `request` is the serialized protobuf request message; the returned bytes # are the serialized response message. fails with a GrpcError on a non-OK @@ -178,7 +212,7 @@ impl Conn: pub fn unary_with_headers_and_deadline(full_method: String, request: Bytes, extra_headers: List[String], timeout_ms: Int) -> Bytes!GrpcError: if not trace.is_active(): framed := frame_message(request) - fields := with_timeout_header(with_metadata(grpc_headers(), extra_headers), timeout_ms) + fields := with_timeout_header(with_metadata(grpc_headers(), self.call_metadata(extra_headers)), timeout_ms) resp := self.client.request_with_timeout("POST", self.authority, full_method, fields, framed, timeout_ms) if resp.is_err: fail transport_error(resp.err) @@ -189,7 +223,7 @@ impl Conn: span.set_attr("rpc.system", "grpc").set_attr("rpc.method", full_method) framed := frame_message(request) started := time.mono_nanos() - fields := with_timeout_header(with_metadata(headers_with_traceparent(span.context()), extra_headers), timeout_ms) + fields := with_timeout_header(with_metadata(headers_with_traceparent(span.context()), self.call_metadata(extra_headers)), timeout_ms) resp := self.client.request_with_timeout("POST", self.authority, full_method, fields, framed, timeout_ms) if resp.is_err: err := transport_error(resp.err) @@ -211,7 +245,7 @@ impl Conn: # close() it when done. pub fn dial(host: String, port: Int) -> Conn!: client := http2.open(host, port)! - return Conn(client: client, authority: host) + return Conn(client: client, authority: host, credentials: no_credentials()) # dial a grpc server using the caller's tls `config` (which must offer alpn # "h2") — for example one built with tls.client_config_with_ca_file to trust a @@ -219,14 +253,14 @@ pub fn dial(host: String, port: Int) -> Conn!: # like dial. pub fn dial_with_config(host: String, port: Int, config: tls.Config) -> Conn!: client := http2.open_with_config(host, port, config)! - return Conn(client: client, authority: host) + return Conn(client: client, authority: host, credentials: no_credentials()) # dial a grpc server over plaintext http/2 (h2c) — no tls. matches a server # started with grpc.serve. use this for local or trusted-network endpoints; for # anything public, dial over tls. otherwise like dial. pub fn dial_h2c(host: String, port: Int) -> Conn!: client := http2.open_h2c(host, port)! - return Conn(client: client, authority: host) + return Conn(client: client, authority: host, credentials: no_credentials()) # make a one-shot unary call: dial host:port, run the method, and close. this is # the simplest entry point; for several calls over one connection, dial() a @@ -1294,6 +1328,139 @@ fn grpc_serve_handler(req: http.HttpRequestBytes) -> http.HttpResponse: framed := frame_message(result.ok) return http.response(200).content_type(GRPC_CONTENT_TYPE).with_body_bytes(framed).with_trailers(grpc_ok_trailers()) +# --- interceptors ------------------------------------------------------------- +# +# an interceptor wraps a dispatch the way a std.web middleware wraps a route +# handler: it takes the rest of the chain (`next`) plus the call, so it can run +# code before or after it, or answer the call itself without invoking next at +# all. compose them with intercept() and hand the result to any serve form — +# the dispatch signature does not change, so a generated router or a +# hand-written dispatch both drop straight in. +# +# fn log_calls(next: fn(String, Bytes) -> Bytes!grpc.GrpcError, path: String, request: Bytes) -> Bytes!grpc.GrpcError: +# print(path) +# return next(path, request) +# +# guarded := grpc.intercept(serve_Chat, [grpc.authorize(check_token), log_calls]) +# grpc.serve("0.0.0.0", 50051, guarded)! +# +# the first interceptor in the list runs outermost, so it sees the call first +# and the reply last — the ordering std.web's use_mw promises. +# +# an interceptor that refuses a call returns refuse(status, message) instead of +# failing directly: a pith lambda can neither `fail` nor use `!`, so the refusal +# is built by a named helper and returned as a value. the ready-made +# interceptors below are shaped that way for the same reason. + +## the refusal an interceptor returns instead of calling next: a failed result +## carrying `status` and `message`, which the server sends back as the call's +## grpc-status and grpc-message. +## +## if not limiter.allow(): +## return grpc.refuse(grpc.GRPC_RESOURCE_EXHAUSTED, "rate limited") +pub fn refuse(status: Int, message: String) -> Bytes!GrpcError: + fail GrpcError(status, message) + +# the pieces the chain runner walks. bundling them in a struct lets the `next` +# closure below capture one struct value and an index rather than the bare +# interceptor and dispatch fn-values — the same reason std.web's Chain exists. +struct Chain: + interceptors: List[fn(fn(String, Bytes) -> Bytes!GrpcError, String, Bytes) -> Bytes!GrpcError] + dispatch: fn(String, Bytes) -> Bytes!GrpcError + +# run the onion from `index` inward: call interceptors[index] with a `next` that +# recurses one level deeper, and when we run past the end, call the dispatch. +# the `next` lambda only forwards the result — it never unwraps one — because a +# lambda cannot use `!`. +fn run_chain(chain: Chain, index: Int, path: String, request: Bytes) -> Bytes!GrpcError: + if index >= chain.interceptors.len(): + return chain.dispatch(path, request)! + interceptor := chain.interceptors[index] + next := fn(p: String, r: Bytes) => run_chain(chain, index + 1, p, r) + return interceptor(next, path, request)! + +## wrap `dispatch` in `interceptors` and return the composed dispatch. the +## first interceptor in the list runs outermost. an empty list returns a +## dispatch that behaves exactly like the one passed in. +pub fn intercept(dispatch: fn(String, Bytes) -> Bytes!GrpcError, interceptors: List[fn(fn(String, Bytes) -> Bytes!GrpcError, String, Bytes) -> Bytes!GrpcError]) -> fn(String, Bytes) -> Bytes!GrpcError: + chain := Chain(interceptors, dispatch) + return fn(path: String, request: Bytes) => run_chain(chain, 0, path, request) + +# what authorize() carries: capturing the verifier inside a struct rather than +# as a bare fn-value keeps the closure on the shape the compiler handles. +struct Authorizer: + verify: fn(Map[String, String]) -> Bool + +## an interceptor that authorizes every call from its metadata. `verify` gets +## the caller's metadata — the headers the client attached, minus the ones grpc +## and the transport own — and returns whether the call may proceed; a refusal +## is UNAUTHENTICATED and never reaches the dispatch. +## +## this is the transport-level hook: it runs before any method body, so one +## interceptor covers every rpc a server exposes. +## +## fn check_token(md: Map[String, String]) -> Bool: +## return jwt.verify_hs256(grpc.bearer_token(md), SECRET).is_ok +## +## grpc.serve(host, port, grpc.intercept(serve_Chat, [grpc.authorize(check_token)]))! +pub fn authorize(verify: fn(Map[String, String]) -> Bool) -> fn(fn(String, Bytes) -> Bytes!GrpcError, String, Bytes) -> Bytes!GrpcError: + guard := Authorizer(verify) + return fn(next: fn(String, Bytes) -> Bytes!GrpcError, path: String, request: Bytes): + if not guard.verify(incoming_metadata()): + return refuse(GRPC_UNAUTHENTICATED, "unauthenticated") + return next(path, request) + +## the bearer token in `metadata`, or "" when the authorization entry is absent +## or is not a bearer credential. the scheme is matched case-insensitively, as +## rfc 7235 requires; the token itself is returned untouched. +pub fn bearer_token(metadata: Map[String, String]) -> String: + value := metadata.get("authorization").unwrap_or("") + if value.len() < 7: + return "" + if value.substring(0, 7).to_lower() != "bearer ": + return "" + return value.substring(7, value.len()) + +## an interceptor that spends one token from `limiter` per call and answers +## RESOURCE_EXHAUSTED when the bucket is empty. build one limiter and share it: +## every serving task drains the same bucket, so the cap holds for the whole +## server no matter how many connections are open. the same limiter can guard +## an http surface through web.rate_limit. +## +## limiter := resilience.rate_limiter(100, 20) +## grpc.serve(host, port, grpc.intercept(serve_Chat, [grpc.rate_limit(limiter)]))! +pub fn rate_limit(limiter: resilience.Limiter) -> fn(fn(String, Bytes) -> Bytes!GrpcError, String, Bytes) -> Bytes!GrpcError: + return fn(next: fn(String, Bytes) -> Bytes!GrpcError, path: String, request: Bytes): + if not limiter.allow(): + return refuse(GRPC_RESOURCE_EXHAUSTED, "rate limited") + return next(path, request) + +## an interceptor that runs calls through a circuit breaker. a dispatch failure +## that is the server's own fault — INTERNAL, UNAVAILABLE, DATA_LOSS, or an +## expired deadline — trips the breaker; a caller's fault (NOT_FOUND, an invalid +## argument, a refused credential) does not, because a client sending bad +## requests is not a reason to stop serving good ones. while the circuit is +## open, callers get UNAVAILABLE immediately instead of waiting on a dependency +## that is already drowning. +## +## breaker := resilience.circuit_breaker(5, 10_000) +## grpc.serve(host, port, grpc.intercept(serve_Chat, [grpc.circuit(breaker)]))! +pub fn circuit(breaker: resilience.Breaker) -> fn(fn(String, Bytes) -> Bytes!GrpcError, String, Bytes) -> Bytes!GrpcError: + return fn(next: fn(String, Bytes) -> Bytes!GrpcError, path: String, request: Bytes): + if not breaker.allow(): + return refuse(GRPC_UNAVAILABLE, "circuit open") + result := next(path, request) + if result.is_err and server_fault(result.err.status): + breaker.failure() + else: + breaker.success() + return result + +## whether `status` names a failure the server is responsible for, and so one +## worth counting against a circuit breaker. a caller's own error is not. +pub fn server_fault(status: Int) -> Bool: + return status == GRPC_INTERNAL or status == GRPC_UNAVAILABLE or status == GRPC_DATA_LOSS or status == GRPC_DEADLINE_EXCEEDED + # serve grpc over plaintext http/2 (h2c). blocks, accepting connections. the # `dispatch` maps a full method path (/pkg.Service/Method) and the request # message bytes to the response message bytes, or a GrpcError. @@ -1603,3 +1770,103 @@ test "only the transient status codes are retryable": assert(not retryable_code(GRPC_DEADLINE_EXCEEDED)) assert(not retryable_code(GRPC_INVALID_ARGUMENT)) assert(not retryable_code(GRPC_UNAUTHENTICATED)) + +# --- interceptor tests -------------------------------------------------------- +# +# the dispatches and interceptors these use are module-level functions because +# a test block cannot declare one; they exist only for the tests below. + +fn echo_dispatch(path: String, request: Bytes) -> Bytes!GrpcError: + if path == "/x.T/Boom": + fail GrpcError(GRPC_INTERNAL, "boom") + if path == "/x.T/BadArg": + fail GrpcError(GRPC_INVALID_ARGUMENT, "bad argument") + return request + +fn tag_a(next: fn(String, Bytes) -> Bytes!GrpcError, path: String, request: Bytes) -> Bytes!GrpcError: + inner := next(path, request)! + return bytes.from_string_utf8("a(" + (inner.to_string_utf8() catch "") + ")") + +fn tag_b(next: fn(String, Bytes) -> Bytes!GrpcError, path: String, request: Bytes) -> Bytes!GrpcError: + inner := next(path, request)! + return bytes.from_string_utf8("b(" + (inner.to_string_utf8() catch "") + ")") + +fn deny_all(next: fn(String, Bytes) -> Bytes!GrpcError, path: String, request: Bytes) -> Bytes!GrpcError: + return refuse(GRPC_PERMISSION_DENIED, "nope") + +test "an empty interceptor list leaves the dispatch alone": + empty: List[fn(fn(String, Bytes) -> Bytes!GrpcError, String, Bytes) -> Bytes!GrpcError] := [] + guarded := intercept(echo_dispatch, empty) + reply := guarded("/x.T/Echo", bytes.from_string_utf8("hi"))! + assert_eq(reply.to_string_utf8()!, "hi") + +test "the first interceptor registered runs outermost": + guarded := intercept(echo_dispatch, [tag_a, tag_b]) + reply := guarded("/x.T/Echo", bytes.from_string_utf8("core"))! + # a wraps b wraps the dispatch, so a's mark is on the outside + assert_eq(reply.to_string_utf8()!, "a(b(core))") + +test "an interceptor can refuse a call without reaching the dispatch": + guarded := intercept(echo_dispatch, [deny_all, tag_b]) + result := guarded("/x.T/Echo", bytes.from_string_utf8("hi")) + assert(result.is_err) + assert_eq(result.err.status, GRPC_PERMISSION_DENIED) + # tag_b never ran, so nothing wrapped the reply + assert_eq(result.err.message, "nope") + +test "a dispatch error propagates out through the interceptors": + guarded := intercept(echo_dispatch, [tag_a]) + result := guarded("/x.T/Boom", bytes.empty()) + assert(result.is_err) + assert_eq(result.err.status, GRPC_INTERNAL) + +test "the rate limit interceptor answers RESOURCE_EXHAUSTED when the bucket is dry": + limiter := resilience.rate_limiter(60, 2) + guarded := intercept(echo_dispatch, [rate_limit(limiter)]) + assert(guarded("/x.T/Echo", bytes.empty()).is_ok) + assert(guarded("/x.T/Echo", bytes.empty()).is_ok) + refused := guarded("/x.T/Echo", bytes.empty()) + assert(refused.is_err) + assert_eq(refused.err.status, GRPC_RESOURCE_EXHAUSTED) + +test "the circuit counts server faults and ignores caller faults": + breaker := resilience.circuit_breaker(2, 60_000) + guarded := intercept(echo_dispatch, [circuit(breaker)]) + # a caller's own error must not trip the breaker, however often it repeats + mut i := 0 + while i < 5: + assert(guarded("/x.T/BadArg", bytes.empty()).is_err) + i = i + 1 + assert(guarded("/x.T/Echo", bytes.empty()).is_ok) + # two server faults reach the threshold and open the circuit + assert(guarded("/x.T/Boom", bytes.empty()).is_err) + assert(guarded("/x.T/Boom", bytes.empty()).is_err) + open := guarded("/x.T/Echo", bytes.empty()) + assert(open.is_err) + assert_eq(open.err.status, GRPC_UNAVAILABLE) + +test "only the server's own failures count against a circuit": + assert(server_fault(GRPC_INTERNAL)) + assert(server_fault(GRPC_UNAVAILABLE)) + assert(server_fault(GRPC_DATA_LOSS)) + assert(server_fault(GRPC_DEADLINE_EXCEEDED)) + assert(not server_fault(GRPC_INVALID_ARGUMENT)) + assert(not server_fault(GRPC_NOT_FOUND)) + assert(not server_fault(GRPC_UNAUTHENTICATED)) + assert(not server_fault(GRPC_OK)) + +test "the bearer token is read case-insensitively, or not at all": + mut md: Map[String, String] := {} + md.insert("authorization", "Bearer abc.def") + assert_eq(bearer_token(md), "abc.def") + # rfc 7235 makes the scheme case-insensitive + md.insert("authorization", "bEaReR xyz") + assert_eq(bearer_token(md), "xyz") + # another scheme is not a bearer credential + md.insert("authorization", "Basic dXNlcjpwdw==") + assert_eq(bearer_token(md), "") + # and neither is a truncated header or none at all + md.insert("authorization", "Bear") + assert_eq(bearer_token(md), "") + mut none_set: Map[String, String] := {} + assert_eq(bearer_token(none_set), "") diff --git a/tests/cases/test_grpc_interceptors.pith b/tests/cases/test_grpc_interceptors.pith new file mode 100644 index 00000000..150c91c5 --- /dev/null +++ b/tests/cases/test_grpc_interceptors.pith @@ -0,0 +1,84 @@ +# end-to-end grpc interceptors: a server composes an authorize interceptor over +# its dispatch, and a real client calls it twice over h2c — once with no +# credentials and once with a bearer token set on the channel. this exercises +# the whole path the unit tests cannot reach: the interceptor runs inside the +# real serve handler, reads metadata the client actually put on the wire, and +# its refusal comes back as a grpc-status the client surfaces as an error. + +import std.net.grpc as grpc +import std.protobuf as protobuf +import std.bytes as bytes +import std.time as time + +PORT := 50079 +TOKEN := "s3cret-token" + +# read protobuf field 1 (a string) from a message. +fn read_field1(data: Bytes) -> String!protobuf.ProtoError: + r := protobuf.reader(data) + mut name := "" + while not r.at_end(): + tag := r.read_tag()! + if tag.field == 1: + name = r.read_string()! + else: + r.skip(tag.wire)! + return name + +fn echo_dispatch(path: String, request: Bytes) -> Bytes!grpc.GrpcError: + if path != "/echo.Echo/Ping": + fail grpc.GrpcError(grpc.GRPC_UNIMPLEMENTED, "no method " + path) + name := read_field1(request) catch "" + w := protobuf.writer() + w.write_string(1, "echo:" + name) catch false + out := w.bytes() + w.free() + return out + +# the transport-level auth check: every method is covered by this one hook. +fn check_token(metadata: Map[String, String]) -> Bool: + return grpc.bearer_token(metadata) == TOKEN + +fn run_server(): + guarded := grpc.intercept(echo_dispatch, [grpc.authorize(check_token)]) + grpc.serve("127.0.0.1", PORT, guarded) catch 0 + +# call once, optionally presenting a credential. returns the reply text, or a +# marker naming how it failed. +fn call_once(present_token: Bool) -> String: + conn_r := grpc.dial_h2c("127.0.0.1", PORT) + if conn_r.is_err: + return "DIAL_ERR" + conn := conn_r.ok + if present_token: + conn.set_credentials(["authorization", "Bearer " + TOKEN]) + w := protobuf.writer() + w.write_string(1, "hello") catch false + reply_r := conn.unary("/echo.Echo/Ping", w.bytes()) + w.free() + if reply_r.is_err: + status := reply_r.err.status + conn.close() + return "status:" + status.to_string() + got := read_field1(reply_r.ok) catch "?" + conn.close() + return got + +# retry until the spawned server is listening; a dial failure is the readiness +# signal, the same shape the other grpc round-trip cases use. +fn call_when_ready(present_token: Bool, timeout_ms: Int) -> String: + mut waited := 0 + while waited < timeout_ms: + got := call_once(present_token) + if got != "DIAL_ERR": + return got + time.delay(25) + waited = waited + 25 + return "DIAL_ERR" + +fn main(): + spawn run_server() + # no credential: the interceptor refuses before the dispatch ever runs + print("anonymous: " + call_when_ready(false, 10000)) + # with the credential: the call reaches the dispatch and comes back + print("credentialed: " + call_when_ready(true, 10000)) diff --git a/tests/expected/test_grpc_interceptors.txt b/tests/expected/test_grpc_interceptors.txt new file mode 100644 index 00000000..7e22e125 --- /dev/null +++ b/tests/expected/test_grpc_interceptors.txt @@ -0,0 +1,2 @@ +anonymous: status:16 +credentialed: echo:hello