Coverage for app/backend/src/couchers/middleware/interceptors.py: 87%
290 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-17 00:57 +0000
1import logging
2from collections.abc import Callable, Mapping
3from dataclasses import dataclass, field
4from datetime import datetime
5from os import getpid
6from threading import get_ident
7from time import perf_counter_ns
8from traceback import format_exception
9from typing import Any, NoReturn, cast
10from zoneinfo import ZoneInfo
12import grpc
13import sentry_sdk
14from google.protobuf.descriptor_pool import DescriptorPool
15from google.protobuf.message import Message
16from opentelemetry import trace
17from sqlalchemy import Function, literal_column, select, update
18from sqlalchemy.dialects.postgresql import insert as pg_insert
19from sqlalchemy.orm import undefer
20from sqlalchemy.sql import func
22from couchers.config import config
23from couchers.constants import (
24 CALL_CANCELLED_ERROR_MESSAGE,
25 COOKIES_AND_AUTH_HEADER_ERROR_MESSAGE,
26 NONEXISTENT_API_CALL_ERROR_MESSAGE,
27 PERMISSION_DENIED_ERROR_MESSAGE,
28 RATE_LIMIT_ERROR_MESSAGE,
29 UNAUTHORIZED_ERROR_MESSAGE,
30 UNKNOWN_ERROR_MESSAGE,
31)
32from couchers.context import CouchersContext, make_interactive_context, make_media_context
33from couchers.db import session_scope
34from couchers.i18n import LocalizationContext
35from couchers.metrics import (
36 observe_api_call,
37 observe_in_servicer_duration_histogram,
38 observe_in_servicer_perf_histograms,
39 observe_in_servicer_pool_wait_histogram,
40 observe_in_servicer_serde_histogram,
41 observe_in_servicer_setup_errors_counter,
42 observe_in_servicer_setup_histogram,
43)
44from couchers.middleware.descriptor_pool import get_descriptor_pool
45from couchers.middleware.errors import CallRejectedError
46from couchers.middleware.perf import PerfResult, read_perf, start_perf
47from couchers.middleware.proto_annotations import find_auth_level
48from couchers.middleware.ratelimit import should_rate_limit
49from couchers.middleware.sanitize import sanitized_bytes
50from couchers.models import APICall, ClientPlatform, User, UserActivity, UserSession
51from couchers.proto import annotations_pb2
52from couchers.proto.annotations_pb2 import AuthLevel
53from couchers.utils import (
54 create_lang_cookie,
55 create_session_cookies,
56 generate_sofa_cookie,
57 parse_api_key,
58 parse_session_cookie,
59 parse_sofa_cookie,
60 parse_ui_lang_cookie,
61 parse_user_id_cookie,
62)
64logger = logging.getLogger(__name__)
66# the prometheus label shared by calls to methods with no servicer registered, whose name is whatever the caller sent
67NONEXISTENT_METHOD_LABEL = "<nonexistent>"
70@dataclass(frozen=True, slots=True, kw_only=True)
71class UserAuthInfo:
72 """Information about an authenticated user session."""
74 user_id: int
75 is_jailed: bool
76 is_editor: bool
77 is_superuser: bool
78 token_expiry: datetime
79 ui_language_preference: str | None
80 timezone: str | None
81 token: str = field(repr=False)
82 is_api_key: bool
85@dataclass(frozen=True, slots=True, kw_only=True)
86class CouchersHeaders:
87 # the user id cookie: client-supplied and unauthenticated, only good for spotting a desynced cookie
88 user_id_str: str | None
89 token: str | None = field(repr=False)
90 sofa: str | None
91 # which mechanism the token came in on, not whether it authenticated: a bad key still reads True here, while
92 # the context's is_api_key is False whenever there's no session at all
93 is_api_key: bool
94 ip_address: str | None
95 user_agent: str | None
96 client_platform: ClientPlatform | None
97 ui_lang: str | None
100def _binned_now() -> Function[Any]:
101 return func.date_bin(
102 literal_column("interval '1 hour'"),
103 func.now(),
104 literal_column("'2000-01-01'::timestamptz"),
105 )
108def _try_get_and_update_user_details(
109 token: str | None,
110 is_api_key: bool,
111 ip_address: str | None,
112 user_agent: str | None,
113 sofa: str | None,
114 client_platform: ClientPlatform | None,
115) -> UserAuthInfo | None:
116 """
117 Tries to get session and user info corresponding to this token.
119 Also updates the user's last active time, token last active time, and increments API call count.
121 Returns UserAuthInfo if a valid session is found, None otherwise.
122 """
123 if not token:
124 return None
126 with session_scope() as session:
127 result = session.execute(
128 select(User, UserSession, User.is_jailed)
129 .select_from(UserSession)
130 .join(User, User.id == UserSession.user_id)
131 .where(User.is_visible)
132 .where(UserSession.token == token)
133 .where(UserSession.is_valid)
134 .where(UserSession.is_api_key == is_api_key)
135 # User.timezone is deferred and read below for every authenticated call, so load it here rather
136 # than paying a second round trip for its ST_Contains against timezone_areas
137 .options(undefer(User.timezone))
138 ).one_or_none()
140 if not result:
141 return None
143 user, user_session, is_jailed = result._tuple()
145 # update user last active time if it's been a while; a non-matching UPDATE takes no row lock, so this
146 # costs nothing on the calls that don't move it
147 touch_user = (
148 update(User)
149 .where(User.id == user.id)
150 .where(User.last_active < func.now() - literal_column("interval '5 minutes'"))
151 .values(last_active=func.now())
152 .cte("touch_user")
153 )
155 # let's update the token
156 touch_session = (
157 update(UserSession)
158 .where(UserSession.token == token)
159 .values(last_seen=func.now(), api_calls=UserSession.api_calls + 1)
160 .cte("touch_session")
161 )
163 # upsert so concurrent requests for the same activity tuple don't race to insert and violate the index
164 insert_stmt = pg_insert(UserActivity).values(
165 user_id=user.id,
166 period=_binned_now(),
167 ip_address=ip_address,
168 user_agent=user_agent,
169 sofa=sofa,
170 client_platform=client_platform,
171 api_calls=1,
172 )
173 # one statement, so the sessions and user_activity row locks that every concurrent call from the same
174 # session queues on are held for a single round trip. postgres leaves the order it applies the CTEs in
175 # undefined, but every caller runs this same statement, so they all take those locks the same way round
176 session.execute(
177 insert_stmt.on_conflict_do_update(
178 index_elements=[
179 UserActivity.user_id,
180 UserActivity.period,
181 UserActivity.ip_address,
182 UserActivity.user_agent,
183 UserActivity.sofa,
184 ],
185 set_={
186 "api_calls": UserActivity.api_calls + 1,
187 "client_platform": func.coalesce(
188 insert_stmt.excluded.client_platform, UserActivity.client_platform
189 ),
190 },
191 ).add_cte(touch_user, touch_session)
192 )
194 # build before committing to avoid expire_on_commit reloading these attributes
195 auth_info = UserAuthInfo(
196 user_id=user.id,
197 is_jailed=is_jailed,
198 is_editor=user.is_editor,
199 is_superuser=user.is_superuser,
200 token_expiry=user_session.expiry,
201 ui_language_preference=user.ui_language_preference,
202 timezone=user.timezone,
203 token=token,
204 is_api_key=is_api_key,
205 )
207 session.commit()
209 return auth_info
212def abort_handler[T, R](
213 message: str,
214 status_code: grpc.StatusCode,
215) -> grpc.RpcMethodHandler[T, R]:
216 def f(request: Any, context: CouchersContext) -> NoReturn:
217 context.abort(status_code, message)
219 return grpc.unary_unary_rpc_method_handler(f)
222def unauthenticated_handler[T, R](
223 message: str = UNAUTHORIZED_ERROR_MESSAGE,
224 status_code: grpc.StatusCode = grpc.StatusCode.UNAUTHENTICATED,
225) -> grpc.RpcMethodHandler[T, R]:
226 return abort_handler(message, status_code)
229def _log_call(
230 *,
231 method: str,
232 status_code: str | None,
233 user_id: int | None,
234 is_api_key: bool,
235 sofa: str | None,
236 headers: CouchersHeaders | None,
237 start: int,
238 perf: PerfResult | None,
239 request: Message | None = None,
240 response: Message | None = None,
241 exception: Exception | None = None,
242 nonexistent_method: bool = False,
243) -> None:
244 """Record a finished call: one api_calls row, plus the per-call Prometheus observations."""
245 duration = (perf_counter_ns() - start) / 1e6 # ms
246 metric_method = NONEXISTENT_METHOD_LABEL if nonexistent_method else method
248 req_bytes = sanitized_bytes(request)
249 res_bytes = sanitized_bytes(response)
250 response_truncated = False
251 truncate_res_bytes_length = 16 * 1024 # 16 kB
252 if res_bytes and len(res_bytes) > truncate_res_bytes_length: 252 ↛ 253line 252 didn't jump to line 253 because the condition on line 252 was never true
253 res_bytes = res_bytes[:truncate_res_bytes_length]
254 response_truncated = True
256 traceback = "".join(format_exception(type(exception), exception, exception.__traceback__)) if exception else None
258 with session_scope() as session:
259 session.add(
260 APICall(
261 is_api_key=is_api_key,
262 method=method,
263 status_code=status_code,
264 duration=duration,
265 user_id=user_id,
266 request=req_bytes,
267 response=res_bytes,
268 response_truncated=response_truncated,
269 traceback=traceback,
270 db_query_count=perf.db_query_count if perf else None,
271 db_write_query_count=perf.db_write_query_count if perf else None,
272 db_time_ms=perf.db_time_ms if perf else None,
273 cpu_ms=perf.cpu_ms if perf else None,
274 client_platform=headers.client_platform if headers else None,
275 ip_address=headers.ip_address if headers else None,
276 user_agent=headers.user_agent if headers else None,
277 sofa=sofa,
278 )
279 )
281 observe_in_servicer_duration_histogram(
282 metric_method, user_id, status_code or "", type(exception).__name__ if exception else "", duration / 1000
283 )
284 observe_api_call(metric_method, headers.client_platform if headers else None)
285 logger.debug(f"{user_id=}, {method=}, {duration=} ms")
288def _log_rejected_call(
289 *,
290 method: str,
291 code: grpc.StatusCode,
292 start: int,
293 handler_call_details: grpc.HandlerCallDetails,
294 user_id: int | None = None,
295 exception: Exception | None = None,
296 nonexistent_method: bool = False,
297) -> None:
298 """Log a call rejected during auth/setup."""
299 try:
300 headers: CouchersHeaders | None = parse_headers(dict(handler_call_details.invocation_metadata))
301 except BadHeaders:
302 headers = None
304 perf = read_perf()
305 _log_call(
306 method=method,
307 status_code=code.name,
308 user_id=user_id,
309 is_api_key=headers.is_api_key if headers else False,
310 sofa=headers.sofa if headers else None,
311 headers=headers,
312 start=start,
313 perf=perf,
314 exception=exception,
315 nonexistent_method=nonexistent_method,
316 )
317 observe_in_servicer_setup_histogram(NONEXISTENT_METHOD_LABEL if nonexistent_method else method, perf)
320def _rejected_call_handler[T, R](
321 *,
322 method: str,
323 message: str,
324 code: grpc.StatusCode,
325 start: int,
326 handler_call_details: grpc.HandlerCallDetails,
327) -> grpc.RpcMethodHandler[T, R]:
328 """Terminate a call that has no handler to run, logging it from the pool thread rather than the serving one."""
330 def f(request: Any, context: grpc.ServicerContext) -> NoReturn:
331 start_perf()
332 _log_rejected_call(
333 method=method,
334 code=code,
335 start=start,
336 handler_call_details=handler_call_details,
337 nonexistent_method=True,
338 )
339 context.abort(code, message)
341 return grpc.unary_unary_rpc_method_handler(f)
344type Cont[T, R] = Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler[T, R] | None]
347@dataclass(frozen=True, slots=True, kw_only=True)
348class AdmittedCall:
349 """What a call that's cleared to run carries into the handler body."""
351 headers: CouchersHeaders
352 auth_info: UserAuthInfo | None
353 sofa: str
354 new_sofa_cookie: str | None
355 localization: LocalizationContext
358@dataclass(frozen=True, slots=True, kw_only=True)
359class RejectedCall:
360 """What a call that didn't clear setup gets terminated and logged with."""
362 code: grpc.StatusCode
363 message: str
364 user_id: int | None
365 # set when setup broke rather than turned the call away, so the caller can report it
366 exception: Exception | None
369def admit_call(pool: DescriptorPool, handler_call_details: grpc.HandlerCallDetails) -> AdmittedCall | RejectedCall:
370 """
371 Pre-RPC setup handling.
373 Never raises: a call that doesn't make it through comes back as a RejectedCall carrying whatever setup had
374 resolved before it stopped, so the caller can log the call it never ran.
375 """
376 auth_info = None
377 try:
378 headers = parse_headers(dict(handler_call_details.invocation_metadata))
380 # if this is not present in prod, it's a Big Bug in config
381 assert config.DEV or headers.ip_address is not None
383 auth_level = find_auth_level(pool, handler_call_details.method)
385 auth_info = _try_get_and_update_user_details(
386 headers.token,
387 headers.is_api_key,
388 headers.ip_address,
389 headers.user_agent,
390 headers.sofa,
391 headers.client_platform,
392 )
394 check_permissions(auth_info, auth_level)
396 if should_rate_limit(handler_call_details.method, headers, auth_info):
397 raise CallRejectedError(RATE_LIMIT_ERROR_MESSAGE, grpc.StatusCode.RESOURCE_EXHAUSTED)
399 if headers.sofa:
400 sofa = headers.sofa
401 new_sofa_cookie = None
402 else:
403 sofa, new_sofa_cookie = generate_sofa_cookie()
405 loc_context = LocalizationContext(
406 locale=(auth_info.ui_language_preference if auth_info else headers.ui_lang) or "",
407 timezone=ZoneInfo((auth_info and auth_info.timezone) or "Etc/UTC"),
408 )
409 except BadHeaders:
410 return RejectedCall(
411 code=grpc.StatusCode.UNAUTHENTICATED,
412 message=COOKIES_AND_AUTH_HEADER_ERROR_MESSAGE,
413 user_id=None,
414 exception=None,
415 )
416 except CallRejectedError as e:
417 return RejectedCall(
418 code=e.code, message=e.msg, user_id=auth_info.user_id if auth_info else None, exception=None
419 )
420 except Exception as e:
421 return RejectedCall(
422 code=grpc.StatusCode.INTERNAL,
423 message=UNKNOWN_ERROR_MESSAGE,
424 user_id=auth_info.user_id if auth_info else None,
425 exception=e,
426 )
428 return AdmittedCall(
429 headers=headers,
430 auth_info=auth_info,
431 sofa=sofa,
432 new_sofa_cookie=new_sofa_cookie,
433 localization=loc_context,
434 )
437class CouchersMiddlewareInterceptor(grpc.ServerInterceptor):
438 """
439 1. Does auth: extracts a session token from a cookie, and authenticates a user with that.
441 Sets context.user_id and context.token if authenticated, otherwise
442 terminates the call with an UNAUTHENTICATED error code.
444 2. Makes sure cookies are in sync.
446 3. Injects a session to get a database transaction.
448 4. Measures and logs the time it takes to service each incoming call.
450 All of that happens in the returned handler, on a thread pool thread. gRPC runs intercept_service inline on the
451 server's single completion-queue thread while holding the server-wide lock, and only submits to the pool once
452 the interceptor chain has returned a handler, so blocking in the interceptor body (the auth query above all)
453 serializes call dispatch for the entire worker process.
454 """
456 def __init__(self) -> None:
457 self._pool = get_descriptor_pool()
459 def intercept_service[T = Message, R = Message](
460 self,
461 continuation: Cont[T, R],
462 handler_call_details: grpc.HandlerCallDetails,
463 ) -> grpc.RpcMethodHandler[T, R]:
464 start = perf_counter_ns()
466 method = handler_call_details.method
468 # only the handler lookup happens here, the rest waits for the handler thread; see the class docstring
469 handler = continuation(handler_call_details)
470 if not handler or not (prev_function := handler.unary_unary):
471 return _rejected_call_handler(
472 method=method,
473 message=NONEXISTENT_API_CALL_ERROR_MESSAGE,
474 code=grpc.StatusCode.UNIMPLEMENTED,
475 start=start,
476 handler_call_details=handler_call_details,
477 )
479 def function_without_couchers_stuff(req: Message, grpc_context: grpc.ServicerContext) -> Message | None:
480 # accounting for the auth/setup phase; the handler re-arms its own below
481 start_perf()
483 call = admit_call(self._pool, handler_call_details)
485 if isinstance(call, RejectedCall):
486 # anything unexpected goes to Sentry before the row below: if the DB is what's broken, that fails too
487 if call.exception:
488 observe_in_servicer_setup_errors_counter(method, type(call.exception).__name__)
489 sentry_sdk.set_tag("context", "servicer_setup")
490 sentry_sdk.set_tag("method", method)
491 sentry_sdk.capture_exception(call.exception)
492 _log_rejected_call(
493 method=method,
494 code=call.code,
495 start=start,
496 handler_call_details=handler_call_details,
497 user_id=call.user_id,
498 exception=call.exception,
499 )
500 grpc_context.abort(call.code, call.message)
502 observe_in_servicer_setup_histogram(method, read_perf())
504 headers = call.headers
505 auth_info = call.auth_info
506 sofa = call.sofa
507 loc_context = call.localization
509 couchers_context = make_interactive_context(
510 grpc_context=grpc_context,
511 user_id=auth_info.user_id if auth_info else None,
512 is_api_key=auth_info.is_api_key if auth_info else False,
513 token=auth_info.token if auth_info else None,
514 localization=loc_context,
515 sofa=sofa,
516 )
518 with session_scope() as session:
519 # force the checkout now so its wait is timed here rather than hiding in the handler's first query
520 pool_wait_start = perf_counter_ns()
521 session.connection()
522 observe_in_servicer_pool_wait_histogram(method, (perf_counter_ns() - pool_wait_start) / 1e9)
523 start_perf()
525 res: Message | None = None
526 exception: Exception | None = None
527 try:
528 _res = prev_function(req, couchers_context, session) # type: ignore[call-arg, arg-type]
529 # flush so pending ORM writes execute (and are counted) before we snapshot; a handler that only
530 # session.add(...)s and returns would otherwise flush at commit, after read_perf()
531 session.flush()
532 res = cast(Message, _res)
533 except Exception as e:
534 exception = e
536 perf = read_perf()
538 if exception and couchers_context._grpc_context:
539 context_code = couchers_context._grpc_context.code() # type: ignore[attr-defined]
540 code = getattr(context_code, "name", None)
541 else:
542 code = None
544 _log_call(
545 method=method,
546 status_code=code,
547 user_id=couchers_context._user_id,
548 is_api_key=cast(bool, couchers_context._is_api_key),
549 sofa=sofa,
550 headers=headers,
551 start=start,
552 perf=perf,
553 request=req,
554 response=res,
555 exception=exception,
556 )
557 observe_in_servicer_perf_histograms(method, perf)
559 if exception:
560 if not code:
561 sentry_sdk.set_tag("context", "servicer")
562 sentry_sdk.set_tag("method", method)
563 sentry_sdk.set_tag("user_agent", headers.user_agent)
564 sentry_sdk.set_tag("ui_lang", loc_context.preferred_locale)
565 sentry_sdk.set_user(
566 {
567 "id": couchers_context._user_id,
568 "ip_address": headers.ip_address,
569 "sofa": sofa[:12],
570 }
571 )
572 sentry_sdk.capture_exception(exception)
574 raise exception
576 if auth_info and not auth_info.is_api_key:
577 # check the two cookies are in sync & that language preference cookie is correct
578 if headers.user_id_str != str(auth_info.user_id): 578 ↛ 582line 578 didn't jump to line 582 because the condition on line 578 was always true
579 couchers_context.set_cookies(
580 create_session_cookies(auth_info.token, auth_info.user_id, auth_info.token_expiry)
581 )
582 if auth_info.ui_language_preference and auth_info.ui_language_preference != headers.ui_lang:
583 couchers_context.set_cookies(create_lang_cookie(auth_info.ui_language_preference))
585 if call.new_sofa_cookie:
586 couchers_context.set_cookies([call.new_sofa_cookie])
588 if not grpc_context.is_active(): 588 ↛ 589line 588 didn't jump to line 589 because the condition on line 588 was never true
589 grpc_context.abort(grpc.StatusCode.INTERNAL, CALL_CANCELLED_ERROR_MESSAGE)
591 couchers_context._send_cookies()
593 return res
595 def timed_serde[A, B](fn: Callable[[A], B], direction: str) -> Callable[[A], B]:
596 def wrapped(arg: A) -> B:
597 t0 = perf_counter_ns()
598 result = fn(arg)
599 observe_in_servicer_serde_histogram(method, direction, (perf_counter_ns() - t0) / 1e9)
600 return result
602 return wrapped
604 # always set for our generated-proto methods, but grpc types them as optional
605 assert handler.request_deserializer is not None and handler.response_serializer is not None
606 return grpc.unary_unary_rpc_method_handler(
607 function_without_couchers_stuff,
608 request_deserializer=timed_serde(handler.request_deserializer, "deserialize"),
609 response_serializer=timed_serde(handler.response_serializer, "serialize"),
610 )
613def parse_headers(headers: Mapping[str, str | bytes]) -> CouchersHeaders:
614 if "cookie" in headers and "authorization" in headers:
615 # for security reasons, only one of "cookie" or "authorization" can be present
616 raise BadHeaders("Both cookies and authorization are present in headers")
617 elif "cookie" in headers:
618 # the session token is passed in cookies, i.e., in the `cookie` header
619 token, is_api_key = parse_session_cookie(headers), False
620 elif "authorization" in headers:
621 # the session token is passed in the `authorization` header
622 token, is_api_key = parse_api_key(headers), True
623 else:
624 # no session found
625 token, is_api_key = None, False
627 ip_address = headers.get("x-couchers-real-ip")
628 user_agent = headers.get("user-agent")
630 # the client (web app or native app) declares its platform via this header
631 client_platform_raw = headers.get("x-couchers-client-platform")
632 client_platform = (
633 ClientPlatform[client_platform_raw]
634 if isinstance(client_platform_raw, str) and client_platform_raw in ClientPlatform.__members__
635 else None
636 )
638 ui_lang = parse_ui_lang_cookie(headers)
639 user_id_str = parse_user_id_cookie(headers)
640 sofa = parse_sofa_cookie(headers)
642 return CouchersHeaders(
643 user_id_str=user_id_str,
644 token=token,
645 sofa=sofa,
646 is_api_key=is_api_key,
647 ip_address=ip_address if isinstance(ip_address, str) else None,
648 user_agent=user_agent if isinstance(user_agent, str) else None,
649 client_platform=client_platform,
650 ui_lang=ui_lang,
651 )
654class BadHeaders(Exception):
655 pass
658def check_permissions(auth_info: UserAuthInfo | None, auth_level: AuthLevel.ValueType) -> None:
659 if not auth_info:
660 # if this isn't an open service, fail
661 if auth_level != annotations_pb2.AUTH_LEVEL_OPEN:
662 raise CallRejectedError(UNAUTHORIZED_ERROR_MESSAGE, grpc.StatusCode.UNAUTHENTICATED)
663 else:
664 # a valid user session was found - check permissions
665 if auth_level == annotations_pb2.AUTH_LEVEL_ADMIN and not auth_info.is_superuser:
666 raise CallRejectedError(PERMISSION_DENIED_ERROR_MESSAGE, grpc.StatusCode.PERMISSION_DENIED)
668 if auth_level == annotations_pb2.AUTH_LEVEL_EDITOR and not auth_info.is_editor:
669 raise CallRejectedError(PERMISSION_DENIED_ERROR_MESSAGE, grpc.StatusCode.PERMISSION_DENIED)
671 # if the user is jailed and this isn't an open or jailed service, fail
672 if auth_info.is_jailed and auth_level not in [
673 annotations_pb2.AUTH_LEVEL_OPEN,
674 annotations_pb2.AUTH_LEVEL_JAILED,
675 ]:
676 raise CallRejectedError(PERMISSION_DENIED_ERROR_MESSAGE, grpc.StatusCode.UNAUTHENTICATED)
679class MediaInterceptor(grpc.ServerInterceptor):
680 """
681 Extracts an "Authorization: Bearer <hex>" header and calls the
682 is_authorized function. Terminates the call with an HTTP error
683 code if not authorized.
685 Also adds a session to called APIs.
686 """
688 def __init__(self, is_authorized: Callable[[str], bool]):
689 self._is_authorized = is_authorized
691 def intercept_service[T, R](
692 self,
693 continuation: Cont[T, R],
694 handler_call_details: grpc.HandlerCallDetails,
695 ) -> grpc.RpcMethodHandler[T, R]:
696 handler = continuation(handler_call_details)
697 if not handler: 697 ↛ 698line 697 didn't jump to line 698 because the condition on line 697 was never true
698 raise RuntimeError("No handler")
700 prev_func = handler.unary_unary
701 if not prev_func: 701 ↛ 702line 701 didn't jump to line 702 because the condition on line 701 was never true
702 raise RuntimeError(f"No prev_function, {handler}")
704 metadata = dict(handler_call_details.invocation_metadata)
706 token = parse_api_key(metadata)
708 if not token or not self._is_authorized(token): 708 ↛ 709line 708 didn't jump to line 709 because the condition on line 708 was never true
709 return unauthenticated_handler()
711 def function_without_session(request: T, grpc_context: grpc.ServicerContext) -> R:
712 with session_scope() as session:
713 return prev_func(request, make_media_context(grpc_context), session) # type: ignore[call-arg, arg-type]
715 return grpc.unary_unary_rpc_method_handler(
716 function_without_session,
717 request_deserializer=handler.request_deserializer,
718 response_serializer=handler.response_serializer,
719 )
722class OTelInterceptor(grpc.ServerInterceptor):
723 """
724 OpenTelemetry tracing
725 """
727 def __init__(self) -> None:
728 self.tracer = trace.get_tracer(__name__)
730 def intercept_service[T, R](
731 self,
732 continuation: Cont[T, R],
733 handler_call_details: grpc.HandlerCallDetails,
734 ) -> grpc.RpcMethodHandler[T, R]:
735 handler = continuation(handler_call_details)
736 if not handler:
737 raise RuntimeError("No handler")
739 prev_func = handler.unary_unary
740 if not prev_func:
741 raise RuntimeError(f"No prev_function, {handler}")
743 method = handler_call_details.method
745 def tracing_function(request: T, context: grpc.ServicerContext) -> R:
746 # method is of the form "/org.couchers.api.core.API/GetUser"
747 _, service_name, method_name = method.split("/")
749 headers = dict(handler_call_details.invocation_metadata)
751 with self.tracer.start_as_current_span("handler") as rollspan:
752 rollspan.set_attribute("rpc.method_full", method)
753 rollspan.set_attribute("rpc.service", service_name)
754 rollspan.set_attribute("rpc.method", method_name)
756 rollspan.set_attribute("rpc.thread", get_ident())
757 rollspan.set_attribute("rpc.pid", getpid())
759 res = prev_func(request, context)
761 rollspan.set_attribute("web.user_agent", headers.get("user-agent") or "")
762 rollspan.set_attribute("web.ip_address", headers.get("x-couchers-real-ip") or "")
764 return res
766 return grpc.unary_unary_rpc_method_handler(
767 tracing_function,
768 request_deserializer=handler.request_deserializer,
769 response_serializer=handler.response_serializer,
770 )
773class ErrorSanitizationInterceptor(grpc.ServerInterceptor):
774 """
775 If the call resulted in a non-gRPC error, this strips away the error details.
777 It's important to put this first, so that it does not interfere with other interceptors.
778 """
780 def intercept_service[T, R](
781 self,
782 continuation: Cont[T, R],
783 handler_call_details: grpc.HandlerCallDetails,
784 ) -> grpc.RpcMethodHandler[T, R]:
785 handler = continuation(handler_call_details)
786 if not handler: 786 ↛ 787line 786 didn't jump to line 787 because the condition on line 786 was never true
787 raise RuntimeError("No handler")
789 prev_func = handler.unary_unary
790 if not prev_func: 790 ↛ 791line 790 didn't jump to line 791 because the condition on line 790 was never true
791 raise RuntimeError(f"No prev_function, {handler}")
793 def sanitizing_function(req: T, context: grpc.ServicerContext) -> R:
794 try:
795 res = prev_func(req, context)
796 except Exception as e:
797 code = context.code() # type: ignore[attr-defined]
798 # the code is one of the RPC error codes if this was failed through abort(), otherwise it's None
799 if not code:
800 logger.exception(e)
801 logger.info("Probably an unknown error! Sanitizing...")
802 context.abort(grpc.StatusCode.INTERNAL, UNKNOWN_ERROR_MESSAGE)
803 else:
804 logger.warning(f"RPC error: {code} in method {handler_call_details.method}")
805 raise e
806 return res
808 return grpc.unary_unary_rpc_method_handler(
809 sanitizing_function,
810 request_deserializer=handler.request_deserializer,
811 response_serializer=handler.response_serializer,
812 )