Coverage for app/backend/src/couchers/interceptors.py: 87%
332 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +0000
1import logging
2from collections.abc import Callable, Mapping
3from copy import deepcopy
4from dataclasses import dataclass, field
5from datetime import datetime, timedelta
6from functools import cache
7from os import getpid
8from threading import get_ident
9from time import perf_counter_ns
10from traceback import format_exception
11from typing import Any, NoReturn, cast, overload
12from zoneinfo import ZoneInfo
14import grpc
15import sentry_sdk
16from google.protobuf.descriptor import Descriptor, ServiceDescriptor
17from google.protobuf.descriptor_pool import DescriptorPool
18from google.protobuf.message import Message
19from opentelemetry import trace
20from sqlalchemy import Function, literal_column, select
21from sqlalchemy.dialects.postgresql import insert as pg_insert
22from sqlalchemy.sql import func
24from couchers.config import config
25from couchers.constants import (
26 CALL_CANCELLED_ERROR_MESSAGE,
27 COOKIES_AND_AUTH_HEADER_ERROR_MESSAGE,
28 MISSING_AUTH_LEVEL_ERROR_MESSAGE,
29 NONEXISTENT_API_CALL_ERROR_MESSAGE,
30 PERMISSION_DENIED_ERROR_MESSAGE,
31 UNAUTHORIZED_ERROR_MESSAGE,
32 UNKNOWN_ERROR_MESSAGE,
33)
34from couchers.context import CouchersContext, make_interactive_context, make_media_context
35from couchers.db import session_scope
36from couchers.descriptor_pool import get_descriptor_pool
37from couchers.i18n import LocalizationContext
38from couchers.metrics import (
39 observe_api_call,
40 observe_in_servicer_duration_histogram,
41 observe_in_servicer_perf_histograms,
42 observe_in_servicer_pool_wait_histogram,
43 observe_in_servicer_serde_histogram,
44 observe_in_servicer_setup_errors_counter,
45 observe_in_servicer_setup_histogram,
46)
47from couchers.models import APICall, ClientPlatform, User, UserActivity, UserSession
48from couchers.perf import PerfResult, read_perf, start_perf
49from couchers.proto import annotations_pb2
50from couchers.proto.annotations_pb2 import AuthLevel
51from couchers.utils import (
52 create_lang_cookie,
53 create_session_cookies,
54 generate_sofa_cookie,
55 now,
56 parse_api_key,
57 parse_session_cookie,
58 parse_sofa_cookie,
59 parse_ui_lang_cookie,
60 parse_user_id_cookie,
61)
63logger = logging.getLogger(__name__)
66@dataclass(frozen=True, slots=True, kw_only=True)
67class UserAuthInfo:
68 """Information about an authenticated user session."""
70 user_id: int
71 is_jailed: bool
72 is_editor: bool
73 is_superuser: bool
74 token_expiry: datetime
75 ui_language_preference: str | None
76 timezone: str | None
77 token: str = field(repr=False)
78 is_api_key: bool
81def _binned_now() -> Function[Any]:
82 return func.date_bin(
83 literal_column("interval '1 hour'"),
84 func.now(),
85 literal_column("'2000-01-01'::timestamptz"),
86 )
89def _try_get_and_update_user_details(
90 token: str | None,
91 is_api_key: bool,
92 ip_address: str | None,
93 user_agent: str | None,
94 sofa: str | None,
95 client_platform: ClientPlatform | None,
96) -> UserAuthInfo | None:
97 """
98 Tries to get session and user info corresponding to this token.
100 Also updates the user's last active time, token last active time, and increments API call count.
102 Returns UserAuthInfo if a valid session is found, None otherwise.
103 """
104 if not token:
105 return None
107 with session_scope() as session:
108 result = session.execute(
109 select(User, UserSession, User.is_jailed)
110 .select_from(UserSession)
111 .join(User, User.id == UserSession.user_id)
112 .where(User.is_visible)
113 .where(UserSession.token == token)
114 .where(UserSession.is_valid)
115 .where(UserSession.is_api_key == is_api_key)
116 ).one_or_none()
118 if not result:
119 return None
121 user, user_session, is_jailed = result._tuple()
123 # update user last active time if it's been a while
124 if now() - user.last_active > timedelta(minutes=5):
125 user.last_active = func.now()
127 # let's update the token
128 user_session.last_seen = func.now()
129 user_session.api_calls += 1
131 # upsert so concurrent requests for the same activity tuple don't race to insert and violate the index
132 insert_stmt = pg_insert(UserActivity).values(
133 user_id=user.id,
134 period=_binned_now(),
135 ip_address=ip_address,
136 user_agent=user_agent,
137 sofa=sofa,
138 client_platform=client_platform,
139 api_calls=1,
140 )
141 session.execute(
142 insert_stmt.on_conflict_do_update(
143 index_elements=[
144 UserActivity.user_id,
145 UserActivity.period,
146 UserActivity.ip_address,
147 UserActivity.user_agent,
148 UserActivity.sofa,
149 ],
150 set_={
151 "api_calls": UserActivity.api_calls + 1,
152 "client_platform": func.coalesce(
153 insert_stmt.excluded.client_platform, UserActivity.client_platform
154 ),
155 },
156 )
157 )
159 # build before committing to avoid expire_on_commit reloading these attributes
160 auth_info = UserAuthInfo(
161 user_id=user.id,
162 is_jailed=is_jailed,
163 is_editor=user.is_editor,
164 is_superuser=user.is_superuser,
165 token_expiry=user_session.expiry,
166 ui_language_preference=user.ui_language_preference,
167 timezone=user.timezone,
168 token=token,
169 is_api_key=is_api_key,
170 )
172 session.commit()
174 return auth_info
177def abort_handler[T, R](
178 message: str,
179 status_code: grpc.StatusCode,
180) -> grpc.RpcMethodHandler[T, R]:
181 def f(request: Any, context: CouchersContext) -> NoReturn:
182 context.abort(status_code, message)
184 return grpc.unary_unary_rpc_method_handler(f)
187def unauthenticated_handler[T, R](
188 message: str = UNAUTHORIZED_ERROR_MESSAGE,
189 status_code: grpc.StatusCode = grpc.StatusCode.UNAUTHENTICATED,
190) -> grpc.RpcMethodHandler[T, R]:
191 return abort_handler(message, status_code)
194@cache
195def _descriptor_has_sensitive(descriptor: Descriptor) -> bool:
196 """Whether this message type transitively contains any field marked sensitive."""
197 seen: set[Descriptor] = set()
198 stack = [descriptor]
199 while stack:
200 d = stack.pop()
201 if d in seen:
202 continue
203 seen.add(d)
204 for f in d.fields:
205 if f.GetOptions().Extensions[annotations_pb2.sensitive]:
206 return True
207 if f.message_type is not None:
208 stack.append(f.message_type)
209 return False
212@dataclass(frozen=True, slots=True)
213class _SanitizePlan:
214 fields_to_clear: tuple[str, ...]
215 fields_to_recurse: tuple[tuple[str, bool], ...] # (field name, is_repeated)
218@cache
219def _sanitize_plan(descriptor: Descriptor) -> _SanitizePlan:
220 """For a message type, the fields to clear and the subfields worth recursing into."""
221 clear = []
222 recurse = []
223 for f in descriptor.fields:
224 if f.GetOptions().Extensions[annotations_pb2.sensitive]:
225 clear.append(f.name)
226 elif f.message_type is not None and _descriptor_has_sensitive(f.message_type):
227 recurse.append((f.name, f.is_repeated))
228 return _SanitizePlan(fields_to_clear=tuple(clear), fields_to_recurse=tuple(recurse))
231def _sanitize_message(message: Message) -> None:
232 plan = _sanitize_plan(message.DESCRIPTOR)
233 for name in plan.fields_to_clear:
234 message.ClearField(name)
235 for name, is_repeated in plan.fields_to_recurse:
236 submessage = getattr(message, name)
237 if not submessage: 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true
238 continue
239 if is_repeated: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 for msg in submessage:
241 _sanitize_message(msg)
242 else:
243 _sanitize_message(submessage)
246@overload
247def _sanitized_bytes(proto: Message) -> bytes: ...
248@overload
249def _sanitized_bytes(proto: None) -> None: ...
250def _sanitized_bytes(proto: Message | None) -> bytes | None:
251 """
252 Remove fields marked sensitive and return serialized bytes.
254 Sensitivity is static per message type, so the descriptor analysis is cached: messages whose type has no
255 sensitive field anywhere serialize directly without a copy or walk.
256 """
257 if not proto:
258 return None
260 if not _descriptor_has_sensitive(proto.DESCRIPTOR):
261 return proto.SerializeToString()
263 new_proto = deepcopy(proto)
264 _sanitize_message(new_proto)
265 return new_proto.SerializeToString()
268def _store_log(
269 *,
270 method: str,
271 status_code: str | None = None,
272 duration: float,
273 user_id: int | None,
274 is_api_key: bool,
275 request: Message,
276 response: Message | None,
277 traceback: str | None = None,
278 perf_report: str | None = None,
279 perf: PerfResult | None = None,
280 client_platform: ClientPlatform | None = None,
281 ip_address: str | None,
282 user_agent: str | None,
283 sofa: str | None,
284) -> None:
285 req_bytes = _sanitized_bytes(request)
286 res_bytes = _sanitized_bytes(response)
287 with session_scope() as session:
288 response_truncated = False
289 truncate_res_bytes_length = 16 * 1024 # 16 kB
290 if res_bytes and len(res_bytes) > truncate_res_bytes_length: 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true
291 res_bytes = res_bytes[:truncate_res_bytes_length]
292 response_truncated = True
293 session.add(
294 APICall(
295 is_api_key=is_api_key,
296 method=method,
297 status_code=status_code,
298 duration=duration,
299 user_id=user_id,
300 request=req_bytes,
301 response=res_bytes,
302 response_truncated=response_truncated,
303 traceback=traceback,
304 perf_report=perf_report,
305 db_query_count=perf.db_query_count if perf else None,
306 db_write_query_count=perf.db_write_query_count if perf else None,
307 db_time_ms=perf.db_time_ms if perf else None,
308 cpu_ms=perf.cpu_ms if perf else None,
309 client_platform=client_platform,
310 ip_address=ip_address,
311 user_agent=user_agent,
312 sofa=sofa,
313 )
314 )
315 logger.debug(f"{user_id=}, {method=}, {duration=} ms")
318type Cont[T, R] = Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler[T, R] | None]
321class CouchersMiddlewareInterceptor(grpc.ServerInterceptor):
322 """
323 1. Does auth: extracts a session token from a cookie, and authenticates a user with that.
325 Sets context.user_id and context.token if authenticated, otherwise
326 terminates the call with an UNAUTHENTICATED error code.
328 2. Makes sure cookies are in sync.
330 3. Injects a session to get a database transaction.
332 4. Measures and logs the time it takes to service each incoming call.
333 """
335 def __init__(self) -> None:
336 self._pool = get_descriptor_pool()
338 def intercept_service[T = Message, R = Message](
339 self,
340 continuation: Cont[T, R],
341 handler_call_details: grpc.HandlerCallDetails,
342 ) -> grpc.RpcMethodHandler[T, R]:
343 start = perf_counter_ns()
345 method = handler_call_details.method
347 # accounting for the auth/setup phase; the handler re-arms its own below
348 start_perf()
350 try:
351 try:
352 auth_level = find_auth_level(self._pool, method)
353 except AbortError as ae:
354 return abort_handler(ae.msg, ae.code)
356 try:
357 headers = parse_headers(dict(handler_call_details.invocation_metadata))
358 except BadHeaders:
359 return unauthenticated_handler(COOKIES_AND_AUTH_HEADER_ERROR_MESSAGE)
361 # if this is not present in prod, it's a Big Bug in config
362 assert config.DEV or headers.ip_address is not None
364 auth_info = _try_get_and_update_user_details(
365 headers.token,
366 headers.is_api_key,
367 headers.ip_address,
368 headers.user_agent,
369 headers.sofa,
370 headers.client_platform,
371 )
373 try:
374 check_permissions(auth_info, auth_level)
375 except AbortError as ae:
376 return unauthenticated_handler(ae.msg, ae.code)
378 if not (handler := continuation(handler_call_details)): 378 ↛ 379line 378 didn't jump to line 379 because the condition on line 378 was never true
379 raise RuntimeError(f"No handler in '{method}'")
381 if not (prev_function := handler.unary_unary): 381 ↛ 382line 381 didn't jump to line 382 because the condition on line 381 was never true
382 raise RuntimeError(f"No prev_function in '{method}', {handler}")
384 if headers.sofa:
385 sofa = headers.sofa
386 new_sofa_cookie = None
387 else:
388 sofa, new_sofa_cookie = generate_sofa_cookie()
390 loc_context = LocalizationContext(
391 locale=(auth_info.ui_language_preference if auth_info else headers.ui_lang) or "",
392 timezone=ZoneInfo((auth_info and auth_info.timezone) or "Etc/UTC"),
393 )
395 observe_in_servicer_setup_histogram(method, read_perf())
396 except Exception as e:
397 observe_in_servicer_setup_errors_counter(method, type(e).__name__)
398 sentry_sdk.set_tag("context", "servicer_setup")
399 sentry_sdk.set_tag("method", method)
400 sentry_sdk.capture_exception(e)
401 return abort_handler(UNKNOWN_ERROR_MESSAGE, grpc.StatusCode.INTERNAL)
403 def function_without_couchers_stuff(req: Message, grpc_context: grpc.ServicerContext) -> Message | None:
404 couchers_context = make_interactive_context(
405 grpc_context=grpc_context,
406 user_id=auth_info.user_id if auth_info else None,
407 is_api_key=auth_info.is_api_key if auth_info else False,
408 token=auth_info.token if auth_info else None,
409 localization=loc_context,
410 sofa=sofa,
411 )
413 with session_scope() as session:
414 # force the checkout now so its wait is timed here rather than hiding in the handler's first query
415 pool_wait_start = perf_counter_ns()
416 session.connection()
417 observe_in_servicer_pool_wait_histogram(method, (perf_counter_ns() - pool_wait_start) / 1e9)
418 start_perf()
419 try:
420 _res = prev_function(req, couchers_context, session) # type: ignore[call-arg, arg-type]
421 res = cast(Message, _res)
422 # flush so pending ORM writes execute (and are counted) before we snapshot; a handler that only
423 # session.add(...)s and returns would otherwise flush at commit, after read_perf()
424 session.flush()
425 perf = read_perf()
426 finished = perf_counter_ns()
427 duration = (finished - start) / 1e6 # ms
428 _store_log(
429 method=method,
430 duration=duration,
431 user_id=couchers_context._user_id,
432 is_api_key=cast(bool, couchers_context._is_api_key),
433 request=req,
434 response=res,
435 perf=perf,
436 client_platform=headers.client_platform,
437 ip_address=headers.ip_address,
438 user_agent=headers.user_agent,
439 sofa=sofa,
440 )
441 observe_in_servicer_duration_histogram(method, couchers_context._user_id, "", "", duration / 1000)
442 observe_api_call(method, headers.client_platform)
443 observe_in_servicer_perf_histograms(method, perf)
444 except Exception as e:
445 perf = read_perf()
446 finished = perf_counter_ns()
447 duration = (finished - start) / 1e6 # ms
449 if couchers_context._grpc_context: 449 ↛ 453line 449 didn't jump to line 453 because the condition on line 449 was always true
450 context_code = couchers_context._grpc_context.code() # type: ignore[attr-defined]
451 code = getattr(context_code, "name", None)
452 else:
453 code = None
455 traceback = "".join(format_exception(type(e), e, e.__traceback__))
456 _store_log(
457 method=method,
458 status_code=code,
459 duration=duration,
460 user_id=couchers_context._user_id,
461 is_api_key=cast(bool, couchers_context._is_api_key),
462 request=req,
463 response=None,
464 traceback=traceback,
465 perf=perf,
466 client_platform=headers.client_platform,
467 ip_address=headers.ip_address,
468 user_agent=headers.user_agent,
469 sofa=sofa,
470 )
471 observe_in_servicer_duration_histogram(
472 method, couchers_context._user_id, code or "", type(e).__name__, duration / 1000
473 )
474 observe_api_call(method, headers.client_platform)
475 observe_in_servicer_perf_histograms(method, perf)
477 if not code:
478 sentry_sdk.set_tag("context", "servicer")
479 sentry_sdk.set_tag("method", method)
480 sentry_sdk.set_tag("user_agent", headers.user_agent)
481 sentry_sdk.set_tag("ui_lang", loc_context.preferred_locale)
482 sentry_sdk.set_user(
483 {
484 "id": couchers_context._user_id,
485 "ip_address": headers.ip_address,
486 "sofa": sofa[:12],
487 }
488 )
489 sentry_sdk.capture_exception(e)
491 raise e
493 if auth_info and not auth_info.is_api_key:
494 # check the two cookies are in sync & that language preference cookie is correct
495 if headers.user_id != str(auth_info.user_id): 495 ↛ 499line 495 didn't jump to line 499 because the condition on line 495 was always true
496 couchers_context.set_cookies(
497 create_session_cookies(auth_info.token, auth_info.user_id, auth_info.token_expiry)
498 )
499 if auth_info.ui_language_preference and auth_info.ui_language_preference != headers.ui_lang:
500 couchers_context.set_cookies(create_lang_cookie(auth_info.ui_language_preference))
502 if new_sofa_cookie:
503 couchers_context.set_cookies([new_sofa_cookie])
505 if not grpc_context.is_active(): 505 ↛ 506line 505 didn't jump to line 506 because the condition on line 505 was never true
506 grpc_context.abort(grpc.StatusCode.INTERNAL, CALL_CANCELLED_ERROR_MESSAGE)
508 couchers_context._send_cookies()
510 return res
512 def timed_serde[A, B](fn: Callable[[A], B], direction: str) -> Callable[[A], B]:
513 def wrapped(arg: A) -> B:
514 t0 = perf_counter_ns()
515 result = fn(arg)
516 observe_in_servicer_serde_histogram(method, direction, (perf_counter_ns() - t0) / 1e9)
517 return result
519 return wrapped
521 # always set for our generated-proto methods, but grpc types them as optional
522 assert handler.request_deserializer is not None and handler.response_serializer is not None
523 return grpc.unary_unary_rpc_method_handler(
524 function_without_couchers_stuff,
525 request_deserializer=timed_serde(handler.request_deserializer, "deserialize"),
526 response_serializer=timed_serde(handler.response_serializer, "serialize"),
527 )
530@dataclass(frozen=True, slots=True, kw_only=True)
531class CouchersHeaders:
532 token: str | None = field(repr=False)
533 is_api_key: bool
534 ip_address: str | None
535 user_agent: str | None
536 client_platform: ClientPlatform | None
537 ui_lang: str | None
538 user_id: str | None
539 sofa: str | None
542def parse_headers(headers: Mapping[str, str | bytes]) -> CouchersHeaders:
543 if "cookie" in headers and "authorization" in headers:
544 # for security reasons, only one of "cookie" or "authorization" can be present
545 raise BadHeaders("Both cookies and authorization are present in headers")
546 elif "cookie" in headers:
547 # the session token is passed in cookies, i.e., in the `cookie` header
548 token, is_api_key = parse_session_cookie(headers), False
549 elif "authorization" in headers:
550 # the session token is passed in the `authorization` header
551 token, is_api_key = parse_api_key(headers), True
552 else:
553 # no session found
554 token, is_api_key = None, False
556 ip_address = headers.get("x-couchers-real-ip")
557 user_agent = headers.get("user-agent")
559 # the client (web app or native app) declares its platform via this header
560 client_platform_raw = headers.get("x-couchers-client-platform")
561 client_platform = (
562 ClientPlatform[client_platform_raw]
563 if isinstance(client_platform_raw, str) and client_platform_raw in ClientPlatform.__members__
564 else None
565 )
567 ui_lang = parse_ui_lang_cookie(headers)
568 user_id = parse_user_id_cookie(headers)
569 sofa = parse_sofa_cookie(headers)
571 return CouchersHeaders(
572 token=token,
573 is_api_key=is_api_key,
574 ip_address=ip_address if isinstance(ip_address, str) else None,
575 user_agent=user_agent if isinstance(user_agent, str) else None,
576 client_platform=client_platform,
577 ui_lang=ui_lang,
578 user_id=user_id,
579 sofa=sofa,
580 )
583class BadHeaders(Exception):
584 pass
587class AbortError(Exception):
588 def __init__(self, msg: str, code: grpc.StatusCode):
589 self.msg = msg
590 self.code = code
593def find_auth_level(pool: DescriptorPool, method: str) -> AuthLevel.ValueType:
594 # method is of the form "/org.couchers.api.core.API/GetUser"
595 _, service_name, method_name = method.split("/")
597 try:
598 service: ServiceDescriptor = pool.FindServiceByName(service_name) # type: ignore[no-untyped-call]
599 service_options = service.GetOptions()
600 except KeyError:
601 raise AbortError(NONEXISTENT_API_CALL_ERROR_MESSAGE, grpc.StatusCode.UNIMPLEMENTED) from None
603 level = service_options.Extensions[annotations_pb2.auth_level]
605 validate_auth_level(level)
607 return level
610def validate_auth_level(auth_level: AuthLevel.ValueType) -> None:
611 # if unknown auth level, then it wasn't set and something's wrong
612 if auth_level == annotations_pb2.AUTH_LEVEL_UNKNOWN:
613 raise AbortError(MISSING_AUTH_LEVEL_ERROR_MESSAGE, grpc.StatusCode.INTERNAL)
615 if auth_level not in { 615 ↛ 622line 615 didn't jump to line 622 because the condition on line 615 was never true
616 annotations_pb2.AUTH_LEVEL_OPEN,
617 annotations_pb2.AUTH_LEVEL_JAILED,
618 annotations_pb2.AUTH_LEVEL_SECURE,
619 annotations_pb2.AUTH_LEVEL_EDITOR,
620 annotations_pb2.AUTH_LEVEL_ADMIN,
621 }:
622 raise AbortError(MISSING_AUTH_LEVEL_ERROR_MESSAGE, grpc.StatusCode.INTERNAL)
625def check_permissions(auth_info: UserAuthInfo | None, auth_level: AuthLevel.ValueType) -> None:
626 if not auth_info:
627 # if this isn't an open service, fail
628 if auth_level != annotations_pb2.AUTH_LEVEL_OPEN:
629 raise AbortError(UNAUTHORIZED_ERROR_MESSAGE, grpc.StatusCode.UNAUTHENTICATED)
630 else:
631 # a valid user session was found - check permissions
632 if auth_level == annotations_pb2.AUTH_LEVEL_ADMIN and not auth_info.is_superuser:
633 raise AbortError(PERMISSION_DENIED_ERROR_MESSAGE, grpc.StatusCode.PERMISSION_DENIED)
635 if auth_level == annotations_pb2.AUTH_LEVEL_EDITOR and not auth_info.is_editor:
636 raise AbortError(PERMISSION_DENIED_ERROR_MESSAGE, grpc.StatusCode.PERMISSION_DENIED)
638 # if the user is jailed and this isn't an open or jailed service, fail
639 if auth_info.is_jailed and auth_level not in [
640 annotations_pb2.AUTH_LEVEL_OPEN,
641 annotations_pb2.AUTH_LEVEL_JAILED,
642 ]:
643 raise AbortError(PERMISSION_DENIED_ERROR_MESSAGE, grpc.StatusCode.UNAUTHENTICATED)
646class MediaInterceptor(grpc.ServerInterceptor):
647 """
648 Extracts an "Authorization: Bearer <hex>" header and calls the
649 is_authorized function. Terminates the call with an HTTP error
650 code if not authorized.
652 Also adds a session to called APIs.
653 """
655 def __init__(self, is_authorized: Callable[[str], bool]):
656 self._is_authorized = is_authorized
658 def intercept_service[T, R](
659 self,
660 continuation: Cont[T, R],
661 handler_call_details: grpc.HandlerCallDetails,
662 ) -> grpc.RpcMethodHandler[T, R]:
663 handler = continuation(handler_call_details)
664 if not handler: 664 ↛ 665line 664 didn't jump to line 665 because the condition on line 664 was never true
665 raise RuntimeError("No handler")
667 prev_func = handler.unary_unary
668 if not prev_func: 668 ↛ 669line 668 didn't jump to line 669 because the condition on line 668 was never true
669 raise RuntimeError(f"No prev_function, {handler}")
671 metadata = dict(handler_call_details.invocation_metadata)
673 token = parse_api_key(metadata)
675 if not token or not self._is_authorized(token): 675 ↛ 676line 675 didn't jump to line 676 because the condition on line 675 was never true
676 return unauthenticated_handler()
678 def function_without_session(request: T, grpc_context: grpc.ServicerContext) -> R:
679 with session_scope() as session:
680 return prev_func(request, make_media_context(grpc_context), session) # type: ignore[call-arg, arg-type]
682 return grpc.unary_unary_rpc_method_handler(
683 function_without_session,
684 request_deserializer=handler.request_deserializer,
685 response_serializer=handler.response_serializer,
686 )
689class OTelInterceptor(grpc.ServerInterceptor):
690 """
691 OpenTelemetry tracing
692 """
694 def __init__(self) -> None:
695 self.tracer = trace.get_tracer(__name__)
697 def intercept_service[T, R](
698 self,
699 continuation: Cont[T, R],
700 handler_call_details: grpc.HandlerCallDetails,
701 ) -> grpc.RpcMethodHandler[T, R]:
702 handler = continuation(handler_call_details)
703 if not handler:
704 raise RuntimeError("No handler")
706 prev_func = handler.unary_unary
707 if not prev_func:
708 raise RuntimeError(f"No prev_function, {handler}")
710 method = handler_call_details.method
712 # method is of the form "/org.couchers.api.core.API/GetUser"
713 _, service_name, method_name = method.split("/")
715 headers = dict(handler_call_details.invocation_metadata)
717 def tracing_function(request: T, context: grpc.ServicerContext) -> R:
718 with self.tracer.start_as_current_span("handler") as rollspan:
719 rollspan.set_attribute("rpc.method_full", method)
720 rollspan.set_attribute("rpc.service", service_name)
721 rollspan.set_attribute("rpc.method", method_name)
723 rollspan.set_attribute("rpc.thread", get_ident())
724 rollspan.set_attribute("rpc.pid", getpid())
726 res = prev_func(request, context)
728 rollspan.set_attribute("web.user_agent", headers.get("user-agent") or "")
729 rollspan.set_attribute("web.ip_address", headers.get("x-couchers-real-ip") or "")
731 return res
733 return grpc.unary_unary_rpc_method_handler(
734 tracing_function,
735 request_deserializer=handler.request_deserializer,
736 response_serializer=handler.response_serializer,
737 )
740class ErrorSanitizationInterceptor(grpc.ServerInterceptor):
741 """
742 If the call resulted in a non-gRPC error, this strips away the error details.
744 It's important to put this first, so that it does not interfere with other interceptors.
745 """
747 def intercept_service[T, R](
748 self,
749 continuation: Cont[T, R],
750 handler_call_details: grpc.HandlerCallDetails,
751 ) -> grpc.RpcMethodHandler[T, R]:
752 handler = continuation(handler_call_details)
753 if not handler: 753 ↛ 754line 753 didn't jump to line 754 because the condition on line 753 was never true
754 raise RuntimeError("No handler")
756 prev_func = handler.unary_unary
757 if not prev_func: 757 ↛ 758line 757 didn't jump to line 758 because the condition on line 757 was never true
758 raise RuntimeError(f"No prev_function, {handler}")
760 def sanitizing_function(req: T, context: grpc.ServicerContext) -> R:
761 try:
762 res = prev_func(req, context)
763 except Exception as e:
764 code = context.code() # type: ignore[attr-defined]
765 # the code is one of the RPC error codes if this was failed through abort(), otherwise it's None
766 if not code:
767 logger.exception(e)
768 logger.info("Probably an unknown error! Sanitizing...")
769 context.abort(grpc.StatusCode.INTERNAL, UNKNOWN_ERROR_MESSAGE)
770 else:
771 logger.warning(f"RPC error: {code} in method {handler_call_details.method}")
772 raise e
773 return res
775 return grpc.unary_unary_rpc_method_handler(
776 sanitizing_function,
777 request_deserializer=handler.request_deserializer,
778 response_serializer=handler.response_serializer,
779 )