grpc interceptors, and channel-wide credentials

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_<Svc>` 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`.

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

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