Coverage for app/backend/src/couchers/interceptors.py: 87%
337 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-16 18:50 +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.orm import undefer
23from sqlalchemy.sql import func
25from couchers.config import config
26from couchers.constants import (
27 CALL_CANCELLED_ERROR_MESSAGE,
28 COOKIES_AND_AUTH_HEADER_ERROR_MESSAGE,
29 MISSING_AUTH_LEVEL_ERROR_MESSAGE,
30 NONEXISTENT_API_CALL_ERROR_MESSAGE,
31 PERMISSION_DENIED_ERROR_MESSAGE,
32 UNAUTHORIZED_ERROR_MESSAGE,
33 UNKNOWN_ERROR_MESSAGE,
34)
35from couchers.context import CouchersContext, make_interactive_context, make_media_context
36from couchers.db import session_scope
37from couchers.descriptor_pool import get_descriptor_pool
38from couchers.i18n import LocalizationContext
39from couchers.metrics import (
40 observe_api_call,
41 observe_in_servicer_duration_histogram,
42 observe_in_servicer_perf_histograms,
43 observe_in_servicer_pool_wait_histogram,
44 observe_in_servicer_serde_histogram,
45 observe_in_servicer_setup_errors_counter,
46 observe_in_servicer_setup_histogram,
47)
48from couchers.models import APICall, ClientPlatform, User, UserActivity, UserSession
49from couchers.perf import PerfResult, read_perf, start_perf
50from couchers.proto import annotations_pb2
51from couchers.proto.annotations_pb2 import AuthLevel
52from couchers.utils import (
53 create_lang_cookie,
54 create_session_cookies,
55 generate_sofa_cookie,
56 now,
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__)
67@dataclass(frozen=True, slots=True, kw_only=True)
68class UserAuthInfo:
69 """Information about an authenticated user session."""
71 user_id: int
72 is_jailed: bool
73 is_editor: bool
74 is_superuser: bool
75 token_expiry: datetime
76 ui_language_preference: str | None
77 timezone: str | None
78 token: str = field(repr=False)
79 is_api_key: bool
82@dataclass(frozen=True, slots=True, kw_only=True)
83class CouchersHeaders:
84 # the user id cookie: client-supplied and unauthenticated, only good for spotting a desynced cookie
85 user_id_str: str | None
86 token: str | None = field(repr=False)
87 sofa: str | None
88 # which mechanism the token came in on, not whether it authenticated: a bad key still reads True here, while
89 # the context's is_api_key is False whenever there's no session at all
90 is_api_key: bool
91 ip_address: str | None
92 user_agent: str | None
93 client_platform: ClientPlatform | None
94 ui_lang: str | None
97def _binned_now() -> Function[Any]:
98 return func.date_bin(
99 literal_column("interval '1 hour'"),
100 func.now(),
101 literal_column("'2000-01-01'::timestamptz"),
102 )
105def _try_get_and_update_user_details(
106 token: str | None,
107 is_api_key: bool,
108 ip_address: str | None,
109 user_agent: str | None,
110 sofa: str | None,
111 client_platform: ClientPlatform | None,
112) -> UserAuthInfo | None:
113 """
114 Tries to get session and user info corresponding to this token.
116 Also updates the user's last active time, token last active time, and increments API call count.
118 Returns UserAuthInfo if a valid session is found, None otherwise.
119 """
120 if not token:
121 return None
123 with session_scope() as session:
124 result = session.execute(
125 select(User, UserSession, User.is_jailed)
126 .select_from(UserSession)
127 .join(User, User.id == UserSession.user_id)
128 .where(User.is_visible)
129 .where(UserSession.token == token)
130 .where(UserSession.is_valid)
131 .where(UserSession.is_api_key == is_api_key)
132 # User.timezone is deferred and read below for every authenticated call, so load it here rather
133 # than paying a second round trip for its ST_Contains against timezone_areas
134 .options(undefer(User.timezone))
135 ).one_or_none()
137 if not result:
138 return None
140 user, user_session, is_jailed = result._tuple()
142 # update user last active time if it's been a while
143 if now() - user.last_active > timedelta(minutes=5):
144 user.last_active = func.now()
146 # let's update the token
147 user_session.last_seen = func.now()
148 user_session.api_calls += 1
150 # upsert so concurrent requests for the same activity tuple don't race to insert and violate the index
151 insert_stmt = pg_insert(UserActivity).values(
152 user_id=user.id,
153 period=_binned_now(),
154 ip_address=ip_address,
155 user_agent=user_agent,
156 sofa=sofa,
157 client_platform=client_platform,
158 api_calls=1,
159 )
160 session.execute(
161 insert_stmt.on_conflict_do_update(
162 index_elements=[
163 UserActivity.user_id,
164 UserActivity.period,
165 UserActivity.ip_address,
166 UserActivity.user_agent,
167 UserActivity.sofa,
168 ],
169 set_={
170 "api_calls": UserActivity.api_calls + 1,
171 "client_platform": func.coalesce(
172 insert_stmt.excluded.client_platform, UserActivity.client_platform
173 ),
174 },
175 )
176 )
178 # build before committing to avoid expire_on_commit reloading these attributes
179 auth_info = UserAuthInfo(
180 user_id=user.id,
181 is_jailed=is_jailed,
182 is_editor=user.is_editor,
183 is_superuser=user.is_superuser,
184 token_expiry=user_session.expiry,
185 ui_language_preference=user.ui_language_preference,
186 timezone=user.timezone,
187 token=token,
188 is_api_key=is_api_key,
189 )
191 session.commit()
193 return auth_info
196def abort_handler[T, R](
197 message: str,
198 status_code: grpc.StatusCode,
199) -> grpc.RpcMethodHandler[T, R]:
200 def f(request: Any, context: CouchersContext) -> NoReturn:
201 context.abort(status_code, message)
203 return grpc.unary_unary_rpc_method_handler(f)
206def unauthenticated_handler[T, R](
207 message: str = UNAUTHORIZED_ERROR_MESSAGE,
208 status_code: grpc.StatusCode = grpc.StatusCode.UNAUTHENTICATED,
209) -> grpc.RpcMethodHandler[T, R]:
210 return abort_handler(message, status_code)
213@cache
214def _descriptor_has_sensitive(descriptor: Descriptor) -> bool:
215 """Whether this message type transitively contains any field marked sensitive."""
216 seen: set[Descriptor] = set()
217 stack = [descriptor]
218 while stack:
219 d = stack.pop()
220 if d in seen:
221 continue
222 seen.add(d)
223 for f in d.fields:
224 if f.GetOptions().Extensions[annotations_pb2.sensitive]:
225 return True
226 if f.message_type is not None:
227 stack.append(f.message_type)
228 return False
231@dataclass(frozen=True, slots=True)
232class _SanitizePlan:
233 fields_to_clear: tuple[str, ...]
234 fields_to_recurse: tuple[tuple[str, bool], ...] # (field name, is_repeated)
237@cache
238def _sanitize_plan(descriptor: Descriptor) -> _SanitizePlan:
239 """For a message type, the fields to clear and the subfields worth recursing into."""
240 clear = []
241 recurse = []
242 for f in descriptor.fields:
243 if f.GetOptions().Extensions[annotations_pb2.sensitive]:
244 clear.append(f.name)
245 elif f.message_type is not None and _descriptor_has_sensitive(f.message_type):
246 recurse.append((f.name, f.is_repeated))
247 return _SanitizePlan(fields_to_clear=tuple(clear), fields_to_recurse=tuple(recurse))
250def _sanitize_message(message: Message) -> None:
251 plan = _sanitize_plan(message.DESCRIPTOR)
252 for name in plan.fields_to_clear:
253 message.ClearField(name)
254 for name, is_repeated in plan.fields_to_recurse:
255 submessage = getattr(message, name)
256 if not submessage: 256 ↛ 257line 256 didn't jump to line 257 because the condition on line 256 was never true
257 continue
258 if is_repeated: 258 ↛ 259line 258 didn't jump to line 259 because the condition on line 258 was never true
259 for msg in submessage:
260 _sanitize_message(msg)
261 else:
262 _sanitize_message(submessage)
265@overload
266def _sanitized_bytes(proto: Message) -> bytes: ...
267@overload
268def _sanitized_bytes(proto: None) -> None: ...
269def _sanitized_bytes(proto: Message | None) -> bytes | None:
270 """
271 Remove fields marked sensitive and return serialized bytes.
273 Sensitivity is static per message type, so the descriptor analysis is cached: messages whose type has no
274 sensitive field anywhere serialize directly without a copy or walk.
275 """
276 if not proto:
277 return None
279 if not _descriptor_has_sensitive(proto.DESCRIPTOR):
280 return proto.SerializeToString()
282 new_proto = deepcopy(proto)
283 _sanitize_message(new_proto)
284 return new_proto.SerializeToString()
287def _log_call(
288 *,
289 method: str,
290 status_code: str | None,
291 user_id: int | None,
292 is_api_key: bool,
293 sofa: str | None,
294 headers: CouchersHeaders,
295 start: int,
296 perf: PerfResult | None,
297 request: Message,
298 response: Message | None,
299 exception: Exception | None,
300) -> None:
301 """Record a finished call: one api_calls row, plus the per-call Prometheus observations."""
302 duration = (perf_counter_ns() - start) / 1e6 # ms
304 req_bytes = _sanitized_bytes(request)
305 res_bytes = _sanitized_bytes(response)
306 response_truncated = False
307 truncate_res_bytes_length = 16 * 1024 # 16 kB
308 if res_bytes and len(res_bytes) > truncate_res_bytes_length: 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true
309 res_bytes = res_bytes[:truncate_res_bytes_length]
310 response_truncated = True
312 traceback = "".join(format_exception(type(exception), exception, exception.__traceback__)) if exception else None
314 with session_scope() as session:
315 session.add(
316 APICall(
317 is_api_key=is_api_key,
318 method=method,
319 status_code=status_code,
320 duration=duration,
321 user_id=user_id,
322 request=req_bytes,
323 response=res_bytes,
324 response_truncated=response_truncated,
325 traceback=traceback,
326 db_query_count=perf.db_query_count if perf else None,
327 db_write_query_count=perf.db_write_query_count if perf else None,
328 db_time_ms=perf.db_time_ms if perf else None,
329 cpu_ms=perf.cpu_ms if perf else None,
330 client_platform=headers.client_platform,
331 ip_address=headers.ip_address,
332 user_agent=headers.user_agent,
333 sofa=sofa,
334 )
335 )
337 observe_in_servicer_duration_histogram(
338 method, user_id, status_code or "", type(exception).__name__ if exception else "", duration / 1000
339 )
340 observe_api_call(method, headers.client_platform)
341 logger.debug(f"{user_id=}, {method=}, {duration=} ms")
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 with."""
362 code: grpc.StatusCode
363 message: str
364 # set when setup broke rather than turned the call away, so the caller can report it
365 exception: Exception | None
368def admit_call(pool: DescriptorPool, handler_call_details: grpc.HandlerCallDetails) -> AdmittedCall | RejectedCall:
369 """
370 Pre-RPC setup handling.
372 Never raises: a call that doesn't make it through comes back as a RejectedCall for the caller to terminate.
373 """
374 try:
375 headers = parse_headers(dict(handler_call_details.invocation_metadata))
377 # if this is not present in prod, it's a Big Bug in config
378 assert config.DEV or headers.ip_address is not None
380 auth_level = find_auth_level(pool, handler_call_details.method)
382 auth_info = _try_get_and_update_user_details(
383 headers.token,
384 headers.is_api_key,
385 headers.ip_address,
386 headers.user_agent,
387 headers.sofa,
388 headers.client_platform,
389 )
391 check_permissions(auth_info, auth_level)
393 if headers.sofa:
394 sofa = headers.sofa
395 new_sofa_cookie = None
396 else:
397 sofa, new_sofa_cookie = generate_sofa_cookie()
399 loc_context = LocalizationContext(
400 locale=(auth_info.ui_language_preference if auth_info else headers.ui_lang) or "",
401 timezone=ZoneInfo((auth_info and auth_info.timezone) or "Etc/UTC"),
402 )
403 except BadHeaders:
404 return RejectedCall(
405 code=grpc.StatusCode.UNAUTHENTICATED, message=COOKIES_AND_AUTH_HEADER_ERROR_MESSAGE, exception=None
406 )
407 except CallRejectedError as e:
408 return RejectedCall(code=e.code, message=e.msg, exception=None)
409 except Exception as e:
410 return RejectedCall(code=grpc.StatusCode.INTERNAL, message=UNKNOWN_ERROR_MESSAGE, exception=e)
412 return AdmittedCall(
413 headers=headers,
414 auth_info=auth_info,
415 sofa=sofa,
416 new_sofa_cookie=new_sofa_cookie,
417 localization=loc_context,
418 )
421class CouchersMiddlewareInterceptor(grpc.ServerInterceptor):
422 """
423 1. Does auth: extracts a session token from a cookie, and authenticates a user with that.
425 Sets context.user_id and context.token if authenticated, otherwise
426 terminates the call with an UNAUTHENTICATED error code.
428 2. Makes sure cookies are in sync.
430 3. Injects a session to get a database transaction.
432 4. Measures and logs the time it takes to service each incoming call.
434 All of that happens in the returned handler, on a thread pool thread. gRPC runs intercept_service inline on the
435 server's single completion-queue thread while holding the server-wide lock, and only submits to the pool once
436 the interceptor chain has returned a handler, so blocking in the interceptor body (the auth query above all)
437 serializes call dispatch for the entire worker process.
438 """
440 def __init__(self) -> None:
441 self._pool = get_descriptor_pool()
443 def intercept_service[T = Message, R = Message](
444 self,
445 continuation: Cont[T, R],
446 handler_call_details: grpc.HandlerCallDetails,
447 ) -> grpc.RpcMethodHandler[T, R]:
448 start = perf_counter_ns()
450 method = handler_call_details.method
452 # only the handler lookup happens here, the rest waits for the handler thread; see the class docstring
453 handler = continuation(handler_call_details)
454 if not handler or not (prev_function := handler.unary_unary): 454 ↛ 455line 454 didn't jump to line 455 because the condition on line 454 was never true
455 return abort_handler(NONEXISTENT_API_CALL_ERROR_MESSAGE, grpc.StatusCode.UNIMPLEMENTED)
457 def function_without_couchers_stuff(req: Message, grpc_context: grpc.ServicerContext) -> Message | None:
458 # accounting for the auth/setup phase; the handler re-arms its own below
459 start_perf()
461 call = admit_call(self._pool, handler_call_details)
463 if isinstance(call, RejectedCall):
464 if call.exception:
465 observe_in_servicer_setup_errors_counter(method, type(call.exception).__name__)
466 sentry_sdk.set_tag("context", "servicer_setup")
467 sentry_sdk.set_tag("method", method)
468 sentry_sdk.capture_exception(call.exception)
469 grpc_context.abort(call.code, call.message)
471 observe_in_servicer_setup_histogram(method, read_perf())
473 headers = call.headers
474 auth_info = call.auth_info
475 sofa = call.sofa
476 loc_context = call.localization
478 couchers_context = make_interactive_context(
479 grpc_context=grpc_context,
480 user_id=auth_info.user_id if auth_info else None,
481 is_api_key=auth_info.is_api_key if auth_info else False,
482 token=auth_info.token if auth_info else None,
483 localization=loc_context,
484 sofa=sofa,
485 )
487 with session_scope() as session:
488 # force the checkout now so its wait is timed here rather than hiding in the handler's first query
489 pool_wait_start = perf_counter_ns()
490 session.connection()
491 observe_in_servicer_pool_wait_histogram(method, (perf_counter_ns() - pool_wait_start) / 1e9)
492 start_perf()
494 res: Message | None = None
495 exception: Exception | None = None
496 try:
497 _res = prev_function(req, couchers_context, session) # type: ignore[call-arg, arg-type]
498 # flush so pending ORM writes execute (and are counted) before we snapshot; a handler that only
499 # session.add(...)s and returns would otherwise flush at commit, after read_perf()
500 session.flush()
501 res = cast(Message, _res)
502 except Exception as e:
503 exception = e
505 perf = read_perf()
507 if exception and couchers_context._grpc_context:
508 context_code = couchers_context._grpc_context.code() # type: ignore[attr-defined]
509 code = getattr(context_code, "name", None)
510 else:
511 code = None
513 _log_call(
514 method=method,
515 status_code=code,
516 user_id=couchers_context._user_id,
517 is_api_key=cast(bool, couchers_context._is_api_key),
518 sofa=sofa,
519 headers=headers,
520 start=start,
521 perf=perf,
522 request=req,
523 response=res,
524 exception=exception,
525 )
526 observe_in_servicer_perf_histograms(method, perf)
528 if exception:
529 if not code:
530 sentry_sdk.set_tag("context", "servicer")
531 sentry_sdk.set_tag("method", method)
532 sentry_sdk.set_tag("user_agent", headers.user_agent)
533 sentry_sdk.set_tag("ui_lang", loc_context.preferred_locale)
534 sentry_sdk.set_user(
535 {
536 "id": couchers_context._user_id,
537 "ip_address": headers.ip_address,
538 "sofa": sofa[:12],
539 }
540 )
541 sentry_sdk.capture_exception(exception)
543 raise exception
545 if auth_info and not auth_info.is_api_key:
546 # check the two cookies are in sync & that language preference cookie is correct
547 if headers.user_id_str != str(auth_info.user_id): 547 ↛ 551line 547 didn't jump to line 551 because the condition on line 547 was always true
548 couchers_context.set_cookies(
549 create_session_cookies(auth_info.token, auth_info.user_id, auth_info.token_expiry)
550 )
551 if auth_info.ui_language_preference and auth_info.ui_language_preference != headers.ui_lang:
552 couchers_context.set_cookies(create_lang_cookie(auth_info.ui_language_preference))
554 if call.new_sofa_cookie:
555 couchers_context.set_cookies([call.new_sofa_cookie])
557 if not grpc_context.is_active(): 557 ↛ 558line 557 didn't jump to line 558 because the condition on line 557 was never true
558 grpc_context.abort(grpc.StatusCode.INTERNAL, CALL_CANCELLED_ERROR_MESSAGE)
560 couchers_context._send_cookies()
562 return res
564 def timed_serde[A, B](fn: Callable[[A], B], direction: str) -> Callable[[A], B]:
565 def wrapped(arg: A) -> B:
566 t0 = perf_counter_ns()
567 result = fn(arg)
568 observe_in_servicer_serde_histogram(method, direction, (perf_counter_ns() - t0) / 1e9)
569 return result
571 return wrapped
573 # always set for our generated-proto methods, but grpc types them as optional
574 assert handler.request_deserializer is not None and handler.response_serializer is not None
575 return grpc.unary_unary_rpc_method_handler(
576 function_without_couchers_stuff,
577 request_deserializer=timed_serde(handler.request_deserializer, "deserialize"),
578 response_serializer=timed_serde(handler.response_serializer, "serialize"),
579 )
582def parse_headers(headers: Mapping[str, str | bytes]) -> CouchersHeaders:
583 if "cookie" in headers and "authorization" in headers:
584 # for security reasons, only one of "cookie" or "authorization" can be present
585 raise BadHeaders("Both cookies and authorization are present in headers")
586 elif "cookie" in headers:
587 # the session token is passed in cookies, i.e., in the `cookie` header
588 token, is_api_key = parse_session_cookie(headers), False
589 elif "authorization" in headers:
590 # the session token is passed in the `authorization` header
591 token, is_api_key = parse_api_key(headers), True
592 else:
593 # no session found
594 token, is_api_key = None, False
596 ip_address = headers.get("x-couchers-real-ip")
597 user_agent = headers.get("user-agent")
599 # the client (web app or native app) declares its platform via this header
600 client_platform_raw = headers.get("x-couchers-client-platform")
601 client_platform = (
602 ClientPlatform[client_platform_raw]
603 if isinstance(client_platform_raw, str) and client_platform_raw in ClientPlatform.__members__
604 else None
605 )
607 ui_lang = parse_ui_lang_cookie(headers)
608 user_id_str = parse_user_id_cookie(headers)
609 sofa = parse_sofa_cookie(headers)
611 return CouchersHeaders(
612 user_id_str=user_id_str,
613 token=token,
614 sofa=sofa,
615 is_api_key=is_api_key,
616 ip_address=ip_address if isinstance(ip_address, str) else None,
617 user_agent=user_agent if isinstance(user_agent, str) else None,
618 client_platform=client_platform,
619 ui_lang=ui_lang,
620 )
623class BadHeaders(Exception):
624 pass
627class CallRejectedError(Exception):
628 def __init__(self, msg: str, code: grpc.StatusCode):
629 self.msg = msg
630 self.code = code
633def find_auth_level(pool: DescriptorPool, method: str) -> AuthLevel.ValueType:
634 # method is of the form "/org.couchers.api.core.API/GetUser"
635 _, service_name, method_name = method.split("/")
637 try:
638 service: ServiceDescriptor = pool.FindServiceByName(service_name) # type: ignore[no-untyped-call]
639 service_options = service.GetOptions()
640 except KeyError:
641 raise CallRejectedError(NONEXISTENT_API_CALL_ERROR_MESSAGE, grpc.StatusCode.UNIMPLEMENTED) from None
643 level = service_options.Extensions[annotations_pb2.auth_level]
645 validate_auth_level(level)
647 return level
650def validate_auth_level(auth_level: AuthLevel.ValueType) -> None:
651 # if unknown auth level, then it wasn't set and something's wrong
652 if auth_level == annotations_pb2.AUTH_LEVEL_UNKNOWN:
653 raise CallRejectedError(MISSING_AUTH_LEVEL_ERROR_MESSAGE, grpc.StatusCode.INTERNAL)
655 if auth_level not in { 655 ↛ 662line 655 didn't jump to line 662 because the condition on line 655 was never true
656 annotations_pb2.AUTH_LEVEL_OPEN,
657 annotations_pb2.AUTH_LEVEL_JAILED,
658 annotations_pb2.AUTH_LEVEL_SECURE,
659 annotations_pb2.AUTH_LEVEL_EDITOR,
660 annotations_pb2.AUTH_LEVEL_ADMIN,
661 }:
662 raise CallRejectedError(MISSING_AUTH_LEVEL_ERROR_MESSAGE, grpc.StatusCode.INTERNAL)
665def check_permissions(auth_info: UserAuthInfo | None, auth_level: AuthLevel.ValueType) -> None:
666 if not auth_info:
667 # if this isn't an open service, fail
668 if auth_level != annotations_pb2.AUTH_LEVEL_OPEN:
669 raise CallRejectedError(UNAUTHORIZED_ERROR_MESSAGE, grpc.StatusCode.UNAUTHENTICATED)
670 else:
671 # a valid user session was found - check permissions
672 if auth_level == annotations_pb2.AUTH_LEVEL_ADMIN and not auth_info.is_superuser:
673 raise CallRejectedError(PERMISSION_DENIED_ERROR_MESSAGE, grpc.StatusCode.PERMISSION_DENIED)
675 if auth_level == annotations_pb2.AUTH_LEVEL_EDITOR and not auth_info.is_editor:
676 raise CallRejectedError(PERMISSION_DENIED_ERROR_MESSAGE, grpc.StatusCode.PERMISSION_DENIED)
678 # if the user is jailed and this isn't an open or jailed service, fail
679 if auth_info.is_jailed and auth_level not in [
680 annotations_pb2.AUTH_LEVEL_OPEN,
681 annotations_pb2.AUTH_LEVEL_JAILED,
682 ]:
683 raise CallRejectedError(PERMISSION_DENIED_ERROR_MESSAGE, grpc.StatusCode.UNAUTHENTICATED)
686class MediaInterceptor(grpc.ServerInterceptor):
687 """
688 Extracts an "Authorization: Bearer <hex>" header and calls the
689 is_authorized function. Terminates the call with an HTTP error
690 code if not authorized.
692 Also adds a session to called APIs.
693 """
695 def __init__(self, is_authorized: Callable[[str], bool]):
696 self._is_authorized = is_authorized
698 def intercept_service[T, R](
699 self,
700 continuation: Cont[T, R],
701 handler_call_details: grpc.HandlerCallDetails,
702 ) -> grpc.RpcMethodHandler[T, R]:
703 handler = continuation(handler_call_details)
704 if not handler: 704 ↛ 705line 704 didn't jump to line 705 because the condition on line 704 was never true
705 raise RuntimeError("No handler")
707 prev_func = handler.unary_unary
708 if not prev_func: 708 ↛ 709line 708 didn't jump to line 709 because the condition on line 708 was never true
709 raise RuntimeError(f"No prev_function, {handler}")
711 metadata = dict(handler_call_details.invocation_metadata)
713 token = parse_api_key(metadata)
715 if not token or not self._is_authorized(token): 715 ↛ 716line 715 didn't jump to line 716 because the condition on line 715 was never true
716 return unauthenticated_handler()
718 def function_without_session(request: T, grpc_context: grpc.ServicerContext) -> R:
719 with session_scope() as session:
720 return prev_func(request, make_media_context(grpc_context), session) # type: ignore[call-arg, arg-type]
722 return grpc.unary_unary_rpc_method_handler(
723 function_without_session,
724 request_deserializer=handler.request_deserializer,
725 response_serializer=handler.response_serializer,
726 )
729class OTelInterceptor(grpc.ServerInterceptor):
730 """
731 OpenTelemetry tracing
732 """
734 def __init__(self) -> None:
735 self.tracer = trace.get_tracer(__name__)
737 def intercept_service[T, R](
738 self,
739 continuation: Cont[T, R],
740 handler_call_details: grpc.HandlerCallDetails,
741 ) -> grpc.RpcMethodHandler[T, R]:
742 handler = continuation(handler_call_details)
743 if not handler:
744 raise RuntimeError("No handler")
746 prev_func = handler.unary_unary
747 if not prev_func:
748 raise RuntimeError(f"No prev_function, {handler}")
750 method = handler_call_details.method
752 def tracing_function(request: T, context: grpc.ServicerContext) -> R:
753 # method is of the form "/org.couchers.api.core.API/GetUser"
754 _, service_name, method_name = method.split("/")
756 headers = dict(handler_call_details.invocation_metadata)
758 with self.tracer.start_as_current_span("handler") as rollspan:
759 rollspan.set_attribute("rpc.method_full", method)
760 rollspan.set_attribute("rpc.service", service_name)
761 rollspan.set_attribute("rpc.method", method_name)
763 rollspan.set_attribute("rpc.thread", get_ident())
764 rollspan.set_attribute("rpc.pid", getpid())
766 res = prev_func(request, context)
768 rollspan.set_attribute("web.user_agent", headers.get("user-agent") or "")
769 rollspan.set_attribute("web.ip_address", headers.get("x-couchers-real-ip") or "")
771 return res
773 return grpc.unary_unary_rpc_method_handler(
774 tracing_function,
775 request_deserializer=handler.request_deserializer,
776 response_serializer=handler.response_serializer,
777 )
780class ErrorSanitizationInterceptor(grpc.ServerInterceptor):
781 """
782 If the call resulted in a non-gRPC error, this strips away the error details.
784 It's important to put this first, so that it does not interfere with other interceptors.
785 """
787 def intercept_service[T, R](
788 self,
789 continuation: Cont[T, R],
790 handler_call_details: grpc.HandlerCallDetails,
791 ) -> grpc.RpcMethodHandler[T, R]:
792 handler = continuation(handler_call_details)
793 if not handler: 793 ↛ 794line 793 didn't jump to line 794 because the condition on line 793 was never true
794 raise RuntimeError("No handler")
796 prev_func = handler.unary_unary
797 if not prev_func: 797 ↛ 798line 797 didn't jump to line 798 because the condition on line 797 was never true
798 raise RuntimeError(f"No prev_function, {handler}")
800 def sanitizing_function(req: T, context: grpc.ServicerContext) -> R:
801 try:
802 res = prev_func(req, context)
803 except Exception as e:
804 code = context.code() # type: ignore[attr-defined]
805 # the code is one of the RPC error codes if this was failed through abort(), otherwise it's None
806 if not code:
807 logger.exception(e)
808 logger.info("Probably an unknown error! Sanitizing...")
809 context.abort(grpc.StatusCode.INTERNAL, UNKNOWN_ERROR_MESSAGE)
810 else:
811 logger.warning(f"RPC error: {code} in method {handler_call_details.method}")
812 raise e
813 return res
815 return grpc.unary_unary_rpc_method_handler(
816 sanitizing_function,
817 request_deserializer=handler.request_deserializer,
818 response_serializer=handler.response_serializer,
819 )