Coverage for app/backend/src/tests/test_interceptors.py: 99%
783 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1from collections.abc import Callable, Generator
2from concurrent import futures
3from contextlib import contextmanager
4from datetime import timedelta
5from threading import current_thread
6from typing import Any
7from unittest.mock import Mock, patch
9import grpc
10import pytest
11from google.protobuf import empty_pb2
12from google.protobuf.descriptor import ServiceDescriptor
13from google.protobuf.descriptor_pool import DescriptorPool
14from sqlalchemy import select, text, update
16from couchers.constants import (
17 COOKIES_AND_AUTH_HEADER_ERROR_MESSAGE,
18 MISSING_AUTH_LEVEL_ERROR_MESSAGE,
19 NONEXISTENT_API_CALL_ERROR_MESSAGE,
20 UNKNOWN_ERROR_MESSAGE,
21)
22from couchers.crypto import b64encode, random_hex, simple_encrypt
23from couchers.db import session_scope
24from couchers.metrics import (
25 api_calls_counter,
26 servicer_db_query_count_histogram,
27 servicer_duration_histogram,
28 servicer_pool_wait_histogram,
29 servicer_serde_histogram,
30 servicer_setup_cpu_time_histogram,
31 servicer_setup_db_time_histogram,
32 servicer_setup_errors_counter,
33)
34from couchers.middleware.errors import CallRejectedError
35from couchers.middleware.interceptors import (
36 NONEXISTENT_METHOD_LABEL,
37 BadHeaders,
38 CouchersMiddlewareInterceptor,
39 ErrorSanitizationInterceptor,
40 UserAuthInfo,
41 _try_get_and_update_user_details,
42 check_permissions,
43 parse_headers,
44)
45from couchers.middleware.proto_annotations import ProtoAnnotations, get_proto_annotations, validate_auth_level
46from couchers.models import APICall, ClientPlatform, User, UserActivity, UserSession
47from couchers.proto import account_pb2, admin_pb2, annotations_pb2, api_pb2, auth_pb2
48from couchers.servicers.account import Account
49from couchers.servicers.api import API
50from couchers.utils import generate_sofa_cookie, now, parse_sofa_cookie
51from tests.fixtures.db import generate_user
52from tests.fixtures.sessions import real_admin_session
55@contextmanager
56def interceptor_dummy_api(
57 rpc,
58 interceptors,
59 service_name="org.couchers.auth.Auth",
60 method_name="SignupFlow",
61 request_type=empty_pb2.Empty,
62 response_type=empty_pb2.Empty,
63 creds=None,
64 call_method_name=None,
65) -> Generator[Callable[..., Any]]:
66 with futures.ThreadPoolExecutor(1) as executor:
67 server = grpc.server(executor, interceptors=interceptors)
68 port = server.add_secure_port("localhost:0", grpc.local_server_credentials())
70 # manually add the handler
71 rpc_method_handlers = {
72 method_name: grpc.unary_unary_rpc_method_handler(
73 rpc,
74 request_deserializer=request_type.FromString,
75 response_serializer=response_type.SerializeToString,
76 )
77 }
78 generic_handler = grpc.method_handlers_generic_handler(service_name, rpc_method_handlers)
79 server.add_generic_rpc_handlers((generic_handler,))
80 server.start()
82 try:
83 with grpc.secure_channel(f"localhost:{port}", creds or grpc.local_channel_credentials()) as channel:
84 yield channel.unary_unary(
85 f"/{service_name}/{call_method_name or method_name}",
86 request_serializer=request_type.SerializeToString,
87 response_deserializer=response_type.FromString,
88 )
89 finally:
90 server.stop(None).wait()
93def _get_histogram_labels_value(method, logged_in, exception, code):
94 metrics = servicer_duration_histogram.collect()
95 servicer_histogram = [m for m in metrics if m.name == "couchers_servicer_duration_seconds"][0]
96 histogram_counts = [
97 s
98 for s in servicer_histogram.samples
99 if s.name == "couchers_servicer_duration_seconds_count"
100 and s.labels["method"] == method
101 and s.labels["logged_in"] == logged_in
102 and s.labels["code"] == code
103 and s.labels["exception"] == exception
104 ]
105 if len(histogram_counts) == 0:
106 return 0
107 return histogram_counts[0].value
110def _get_setup_errors_value(method, exception):
111 metrics = servicer_setup_errors_counter.collect()
112 counter = [m for m in metrics if m.name == "couchers_servicer_setup_errors"][0]
113 samples = [
114 s
115 for s in counter.samples
116 if s.name == "couchers_servicer_setup_errors_total"
117 and s.labels["method"] == method
118 and s.labels["exception"] == exception
119 ]
120 if len(samples) == 0:
121 return 0
122 return samples[0].value
125def test_logging_interceptor_ok():
126 def TestRpc(request, context):
127 return empty_pb2.Empty()
129 with interceptor_dummy_api(TestRpc, interceptors=[ErrorSanitizationInterceptor()]) as call_rpc:
130 call_rpc(empty_pb2.Empty())
133def test_logging_interceptor_all_ignored():
134 # error codes that should not be touched by the interceptor
135 pass_through_status_codes = [
136 # we can't abort with OK
137 # grpc.StatusCode.OK,
138 grpc.StatusCode.CANCELLED,
139 grpc.StatusCode.UNKNOWN,
140 grpc.StatusCode.INVALID_ARGUMENT,
141 grpc.StatusCode.DEADLINE_EXCEEDED,
142 grpc.StatusCode.NOT_FOUND,
143 grpc.StatusCode.ALREADY_EXISTS,
144 grpc.StatusCode.PERMISSION_DENIED,
145 grpc.StatusCode.UNAUTHENTICATED,
146 grpc.StatusCode.RESOURCE_EXHAUSTED,
147 grpc.StatusCode.FAILED_PRECONDITION,
148 grpc.StatusCode.ABORTED,
149 grpc.StatusCode.OUT_OF_RANGE,
150 grpc.StatusCode.UNIMPLEMENTED,
151 grpc.StatusCode.INTERNAL,
152 grpc.StatusCode.UNAVAILABLE,
153 grpc.StatusCode.DATA_LOSS,
154 ]
156 for status_code in pass_through_status_codes:
157 message = random_hex()
159 def TestRpc(request, context):
160 context.abort(status_code, message) # noqa: B023
162 with interceptor_dummy_api(TestRpc, interceptors=[ErrorSanitizationInterceptor()]) as call_rpc:
163 with pytest.raises(grpc.RpcError) as e:
164 call_rpc(empty_pb2.Empty())
165 assert e.value.code() == status_code
166 assert e.value.details() == message
169def test_logging_interceptor_assertion():
170 def TestRpc(request, context):
171 raise AssertionError()
173 with interceptor_dummy_api(TestRpc, interceptors=[ErrorSanitizationInterceptor()]) as call_rpc:
174 with pytest.raises(grpc.RpcError) as e:
175 call_rpc(empty_pb2.Empty())
176 assert e.value.code() == grpc.StatusCode.INTERNAL
177 assert e.value.details() == "An unknown backend error occurred. Please consider filing a bug!"
180def test_logging_interceptor_div0():
181 def TestRpc(request, context):
182 1 / 0 # noqa: B018
184 with interceptor_dummy_api(TestRpc, interceptors=[ErrorSanitizationInterceptor()]) as call_rpc:
185 with pytest.raises(grpc.RpcError) as e:
186 call_rpc(empty_pb2.Empty())
187 assert e.value.code() == grpc.StatusCode.INTERNAL
188 assert e.value.details() == "An unknown backend error occurred. Please consider filing a bug!"
191def test_logging_interceptor_raise():
192 def TestRpc(request, context):
193 raise Exception()
195 with interceptor_dummy_api(TestRpc, interceptors=[ErrorSanitizationInterceptor()]) as call_rpc:
196 with pytest.raises(grpc.RpcError) as e:
197 call_rpc(empty_pb2.Empty())
198 assert e.value.code() == grpc.StatusCode.INTERNAL
199 assert e.value.details() == "An unknown backend error occurred. Please consider filing a bug!"
202def test_logging_interceptor_raise_custom():
203 class _TestingException(Exception):
204 pass
206 def TestRpc(request, context):
207 raise _TestingException("This is a custom exception")
209 with interceptor_dummy_api(TestRpc, interceptors=[ErrorSanitizationInterceptor()]) as call_rpc:
210 with pytest.raises(grpc.RpcError) as e:
211 call_rpc(empty_pb2.Empty())
212 assert e.value.code() == grpc.StatusCode.INTERNAL
213 assert e.value.details() == "An unknown backend error occurred. Please consider filing a bug!"
216def test_tracing_interceptor_ok_open(db):
217 val = _get_histogram_labels_value("/org.couchers.auth.Auth/SignupFlow", "False", "", "")
219 def TestRpc(request, context, session):
220 return empty_pb2.Empty()
222 with interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc:
223 call_rpc(empty_pb2.Empty())
225 with session_scope() as session:
226 trace = session.execute(select(APICall)).scalar_one()
227 assert trace.method == "/org.couchers.auth.Auth/SignupFlow"
228 assert not trace.status_code
229 assert not trace.user_id
230 assert trace.request is not None
231 assert len(trace.request) == 0
232 assert trace.response is not None
233 assert len(trace.response) == 0
234 assert not trace.traceback
236 assert _get_histogram_labels_value("/org.couchers.auth.Auth/SignupFlow", "False", "", "") == val + 1
239def _get_db_query_count_histogram(method):
240 return sum(
241 s.value
242 for m in servicer_db_query_count_histogram.collect()
243 for s in m.samples
244 if s.name == "couchers_servicer_db_query_count_count" and s.labels.get("method") == method
245 )
248def _get_api_call_count(method, platform):
249 return sum(
250 s.value
251 for m in api_calls_counter.collect()
252 for s in m.samples
253 if s.name == "couchers_api_calls_total"
254 and s.labels.get("method") == method
255 and s.labels.get("platform") == platform
256 )
259def test_tracing_interceptor_perf_accounting(db):
260 method = "/org.couchers.auth.Auth/SignupFlow"
261 hist_count_before = _get_db_query_count_histogram(method)
262 api_call_count_before = _get_api_call_count(method, "web_mobile")
264 # handler runs a known number of statements: three reads and one compiled write. The write matches zero rows so
265 # it's side-effect free.
266 def TestRpc(request, context, session):
267 for _ in range(3):
268 session.execute(text("SELECT 1"))
269 session.execute(update(APICall).where(APICall.id == -1).values(method="x"))
270 return empty_pb2.Empty()
272 with interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc:
273 call_rpc(empty_pb2.Empty(), metadata=(("x-couchers-client-platform", "web_mobile"),))
275 with session_scope() as session:
276 trace = session.execute(select(APICall)).scalar_one()
277 assert trace.db_query_count == 4
278 assert trace.db_write_query_count == 1
279 assert trace.db_time_ms is not None and trace.db_time_ms >= 0
280 assert trace.cpu_ms is not None and trace.cpu_ms >= 0
281 # the handler's DB work can't exceed the whole-request wall time
282 assert trace.db_time_ms <= trace.duration
283 assert trace.client_platform == ClientPlatform.web_mobile
285 # the call was also observed into the Prometheus per-request resource histograms and the per-platform call counter
286 assert _get_db_query_count_histogram(method) == hist_count_before + 1
287 assert _get_api_call_count(method, "web_mobile") == api_call_count_before + 1
290def _get_histogram_count(histogram, count_name, **labels):
291 return sum(
292 s.value
293 for m in histogram.collect()
294 for s in m.samples
295 if s.name == count_name and all(s.labels.get(k) == v for k, v in labels.items())
296 )
299def test_tracing_interceptor_phase_histograms(db):
300 # setup db/cpu, pool-wait, and de/serialization are each observed once per call into their own histogram
301 method = "/org.couchers.auth.Auth/SignupFlow"
302 setup_db_before = _get_histogram_count(
303 servicer_setup_db_time_histogram, "couchers_servicer_setup_db_time_seconds_count", method=method
304 )
305 setup_cpu_before = _get_histogram_count(
306 servicer_setup_cpu_time_histogram, "couchers_servicer_setup_cpu_seconds_count", method=method
307 )
308 pool_wait_before = _get_histogram_count(
309 servicer_pool_wait_histogram, "couchers_servicer_pool_wait_seconds_count", method=method
310 )
311 deserialize_before = _get_histogram_count(
312 servicer_serde_histogram, "couchers_servicer_serde_seconds_count", method=method, direction="deserialize"
313 )
314 serialize_before = _get_histogram_count(
315 servicer_serde_histogram, "couchers_servicer_serde_seconds_count", method=method, direction="serialize"
316 )
318 def TestRpc(request, context, session):
319 return empty_pb2.Empty()
321 with interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc:
322 call_rpc(empty_pb2.Empty())
324 assert (
325 _get_histogram_count(
326 servicer_setup_db_time_histogram, "couchers_servicer_setup_db_time_seconds_count", method=method
327 )
328 == setup_db_before + 1
329 )
330 assert (
331 _get_histogram_count(
332 servicer_setup_cpu_time_histogram, "couchers_servicer_setup_cpu_seconds_count", method=method
333 )
334 == setup_cpu_before + 1
335 )
336 assert (
337 _get_histogram_count(servicer_pool_wait_histogram, "couchers_servicer_pool_wait_seconds_count", method=method)
338 == pool_wait_before + 1
339 )
340 assert (
341 _get_histogram_count(
342 servicer_serde_histogram, "couchers_servicer_serde_seconds_count", method=method, direction="deserialize"
343 )
344 == deserialize_before + 1
345 )
346 assert (
347 _get_histogram_count(
348 servicer_serde_histogram, "couchers_servicer_serde_seconds_count", method=method, direction="serialize"
349 )
350 == serialize_before + 1
351 )
354def test_auth_runs_on_the_handler_thread(db):
355 # gRPC runs intercept_service inline on the server's single polling thread, under the server-wide lock, so the auth
356 # query has to happen in the returned handler or it serializes dispatch for every other call in the process
357 threads: dict[str, str] = {}
359 def TestRpc(request, context, session):
360 threads["handler"] = current_thread().name
361 return empty_pb2.Empty()
363 def record_thread(*args, **kwargs):
364 threads["auth"] = current_thread().name
365 return _try_get_and_update_user_details(*args, **kwargs)
367 with (
368 patch("couchers.middleware.interceptors._try_get_and_update_user_details", record_thread),
369 interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc,
370 ):
371 call_rpc(empty_pb2.Empty())
373 assert threads["auth"] == threads["handler"]
376def test_tracing_interceptor_perf_accounting_orm_write(db):
377 # a handler that only session.add(...)s and returns: the INSERT flushes at commit, after read_perf(), so without
378 # the interceptor's explicit flush it would be missed from the write/query counts
379 method = "/org.couchers.auth.Auth/SignupFlow"
381 def TestRpc(request, context, session):
382 session.add(APICall(method="handler-insert", duration=0.0, is_api_key=False, response_truncated=False))
383 return empty_pb2.Empty()
385 with interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc:
386 call_rpc(empty_pb2.Empty())
388 with session_scope() as session:
389 log = session.execute(select(APICall).where(APICall.method == method)).scalar_one()
390 assert log.db_query_count == 1
391 assert log.db_write_query_count == 1
394def test_tracing_interceptor_sensitive(db):
395 val = _get_histogram_labels_value("/org.couchers.auth.Auth/SignupFlow", "False", "", "")
397 def TestRpc(request, context, session):
398 return auth_pb2.AuthReq(user="this is not secret", password="this is secret")
400 with interceptor_dummy_api(
401 TestRpc,
402 interceptors=[CouchersMiddlewareInterceptor()],
403 request_type=auth_pb2.SignupFlowReq,
404 response_type=auth_pb2.AuthReq,
405 ) as call_rpc:
406 call_rpc(
407 auth_pb2.SignupFlowReq(account=auth_pb2.SignupAccount(password="should be removed", username="not removed"))
408 )
410 with session_scope() as session:
411 trace = session.execute(select(APICall)).scalar_one()
412 assert trace.method == "/org.couchers.auth.Auth/SignupFlow"
413 assert not trace.status_code
414 assert not trace.user_id
415 assert not trace.traceback
416 assert trace.request is not None
417 req = auth_pb2.SignupFlowReq.FromString(trace.request)
418 assert not req.account.password
419 assert req.account.username == "not removed"
420 assert trace.response
421 res = auth_pb2.AuthReq.FromString(trace.response)
422 assert res.user == "this is not secret"
423 assert not res.password
425 assert _get_histogram_labels_value("/org.couchers.auth.Auth/SignupFlow", "False", "", "") == val + 1
428def test_tracing_interceptor_sensitive_ping(db):
429 user, token = generate_user()
431 with interceptor_dummy_api(
432 API().GetUser,
433 interceptors=[CouchersMiddlewareInterceptor()],
434 request_type=api_pb2.GetUserReq,
435 response_type=api_pb2.User,
436 service_name="org.couchers.api.core.API",
437 method_name="GetUser",
438 ) as call_rpc:
439 call_rpc(api_pb2.GetUserReq(user=user.username), metadata=(("cookie", f"couchers-sesh={token}"),))
442def test_tracing_interceptor_exception(db):
443 val = _get_histogram_labels_value("/org.couchers.auth.Auth/SignupFlow", "False", "Exception", "")
445 def TestRpc(request, context, session):
446 raise Exception("Some error message")
448 with interceptor_dummy_api(
449 TestRpc,
450 interceptors=[CouchersMiddlewareInterceptor()],
451 request_type=auth_pb2.SignupAccount,
452 response_type=auth_pb2.AuthReq,
453 ) as call_rpc:
454 with pytest.raises(Exception, match="Some error message"):
455 call_rpc(auth_pb2.SignupAccount(password="should be removed", username="not removed"))
457 with session_scope() as session:
458 trace = session.execute(select(APICall)).scalar_one()
459 assert trace.method == "/org.couchers.auth.Auth/SignupFlow"
460 assert not trace.status_code
461 assert not trace.user_id
462 assert trace.traceback
463 assert "Some error message" in trace.traceback
464 assert trace.request is not None
465 req = auth_pb2.SignupAccount.FromString(trace.request)
466 assert not req.password
467 assert req.username == "not removed"
468 assert not trace.response
470 assert _get_histogram_labels_value("/org.couchers.auth.Auth/SignupFlow", "False", "Exception", "") == val + 1
473def test_setup_phase_exception_observed(db):
474 method = "/org.couchers.auth.Auth/SignupFlow"
475 val = _get_setup_errors_value(method, "ValueError")
477 def TestRpc(request, context, session):
478 return empty_pb2.Empty()
480 with (
481 patch("couchers.middleware.interceptors.LocalizationContext", side_effect=ValueError("expected only letters")),
482 patch("couchers.middleware.interceptors.sentry_sdk") as mock_sentry,
483 interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc,
484 ):
485 with pytest.raises(grpc.RpcError) as e:
486 call_rpc(empty_pb2.Empty())
487 assert e.value.code() == grpc.StatusCode.INTERNAL
488 assert e.value.details() == UNKNOWN_ERROR_MESSAGE
489 mock_sentry.capture_exception.assert_called_once()
491 assert _get_setup_errors_value(method, "ValueError") == val + 1
493 with session_scope() as session:
494 trace = session.execute(select(APICall)).scalar_one()
495 assert trace.method == method
496 assert trace.status_code == "INTERNAL"
497 assert trace.traceback
498 assert "expected only letters" in trace.traceback
501def test_tracing_interceptor_abort(db):
502 val = _get_histogram_labels_value("/org.couchers.auth.Auth/SignupFlow", "False", "Exception", "FAILED_PRECONDITION")
504 def TestRpc(request, context, session):
505 context.abort(grpc.StatusCode.FAILED_PRECONDITION, "now a grpc abort")
507 with interceptor_dummy_api(
508 TestRpc,
509 interceptors=[CouchersMiddlewareInterceptor()],
510 request_type=auth_pb2.SignupAccount,
511 response_type=auth_pb2.AuthReq,
512 ) as call_rpc:
513 with pytest.raises(Exception, match="now a grpc abort"):
514 call_rpc(auth_pb2.SignupAccount(password="should be removed", username="not removed"))
516 with session_scope() as session:
517 trace = session.execute(select(APICall)).scalar_one()
518 assert trace.method == "/org.couchers.auth.Auth/SignupFlow"
519 assert trace.status_code == "FAILED_PRECONDITION"
520 assert not trace.user_id
521 assert trace.traceback
522 assert "now a grpc abort" in trace.traceback
523 assert trace.request is not None
524 req = auth_pb2.SignupAccount.FromString(trace.request)
525 assert not req.password
526 assert req.username == "not removed"
527 assert not trace.response
529 assert (
530 _get_histogram_labels_value("/org.couchers.auth.Auth/SignupFlow", "False", "Exception", "FAILED_PRECONDITION")
531 == val + 1
532 )
535def cookie_auth(token: str) -> tuple[str, str]:
536 return "cookie", f"couchers-sesh={token}"
539def api_auth(token: str) -> tuple[str, str]:
540 return "authorization", f"Bearer {token}"
543def test_auth_interceptor(db):
544 super_user, super_token = generate_user(is_superuser=True)
545 user, token = generate_user()
546 deleted_user, deleted_token = generate_user(delete_user=True)
548 with real_admin_session(super_token) as api:
549 api.CreateApiKey(admin_pb2.CreateApiKeyReq(user=user.username))
551 with session_scope() as session:
552 api_key = session.execute(select(UserSession.token).where(UserSession.is_api_key)).scalar_one()
554 account = Account()
556 rpc_def = {
557 "rpc": account.GetAccountInfo,
558 "service_name": "org.couchers.api.account.Account",
559 "method_name": "GetAccountInfo",
560 "interceptors": [CouchersMiddlewareInterceptor()],
561 "request_type": empty_pb2.Empty,
562 "response_type": account_pb2.GetAccountInfoRes,
563 }
565 # no creds, no-go for secure APIs
566 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
567 with pytest.raises(grpc.RpcError) as e:
568 call_rpc(empty_pb2.Empty())
569 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED
570 assert e.value.details() == "Unauthorized"
572 # can auth with cookie
573 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
574 res1 = call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token),))
575 assert res1.username == user.username
577 with session_scope() as session:
578 api_calls = session.execute(select(UserActivity.api_calls).where(UserActivity.user_id == user.id)).scalar_one()
579 assert api_calls == 1
581 # can't auth with a wrong cookie
582 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
583 with pytest.raises(grpc.RpcError) as e:
584 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(random_hex(32)),))
585 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED
586 assert e.value.details() == "Unauthorized"
588 # can auth with an api key
589 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
590 res2 = call_rpc(empty_pb2.Empty(), metadata=(api_auth(api_key),))
591 assert res2.username == user.username
593 with session_scope() as session:
594 api_calls = session.execute(select(UserActivity.api_calls).where(UserActivity.user_id == user.id)).scalar_one()
595 assert api_calls == 2
597 # can't auth with a wrong api key
598 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
599 with pytest.raises(grpc.RpcError) as e:
600 call_rpc(empty_pb2.Empty(), metadata=(api_auth(random_hex(32)),))
601 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED
602 assert e.value.details() == "Unauthorized"
604 # can auth with grpc helper (they do the same as above)
605 comp_creds = grpc.composite_channel_credentials(
606 grpc.local_channel_credentials(), grpc.access_token_call_credentials(api_key)
607 )
608 with interceptor_dummy_api(**rpc_def, creds=comp_creds) as call_rpc:
609 res3 = call_rpc(empty_pb2.Empty())
610 assert res3.username == user.username
612 # can't auth with both
613 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
614 with pytest.raises(grpc.RpcError) as e:
615 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token), api_auth(api_key)))
616 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED
617 assert e.value.details() == 'Both "cookie" and "authorization" in request'
619 # malformed bearer
620 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
621 with pytest.raises(grpc.RpcError) as e:
622 call_rpc(empty_pb2.Empty(), metadata=(("authorization", f"bearer {api_key}"),))
623 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED
624 assert e.value.details() == "Unauthorized"
626 # Invisible (deleted) user
627 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
628 with pytest.raises(grpc.RpcError) as e:
629 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(deleted_token),))
630 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED
631 assert e.value.details() == "Unauthorized"
633 # Invalid (expired) session
634 long_ago = now() - timedelta(weeks=100)
635 with session_scope() as session:
636 session.execute(update(UserSession).values(last_seen=long_ago).where(UserSession.token == token))
638 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
639 with pytest.raises(grpc.RpcError) as e:
640 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token),))
641 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED
642 assert e.value.details() == "Unauthorized"
644 # API key token, but session is for session cookie (probably impossible, but...)
645 with session_scope() as session:
646 session.execute(update(UserSession).values(last_seen=now(), is_api_key=True).where(UserSession.token == token))
648 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
649 with pytest.raises(grpc.RpcError) as e:
650 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token),))
651 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED
652 assert e.value.details() == "Unauthorized"
654 # Check that metadata are updated
655 six_minutes_ago = now() - timedelta(minutes=6)
656 with session_scope() as session:
657 # Return the session to normal
658 user_session = session.execute(select(UserSession).where(UserSession.token == token)).scalar_one()
659 user_session.is_api_key = False
660 api_calls = user_session.api_calls
662 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
663 res4 = call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token),))
664 assert res4.username == user.username
666 with session_scope() as session:
667 user_session = session.execute(select(UserSession).where(UserSession.token == token)).scalar_one()
668 assert user_session.api_calls == api_calls + 1
669 assert user_session.last_seen > now() - timedelta(seconds=1)
671 # Simulate user inactivity, so last_active is updated on the next api call.
672 session.execute(update(User).values(last_active=six_minutes_ago).where(User.id == user.id))
674 # Check that last_active is updated if it wasn't updated in a while.
675 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
676 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token),))
678 with session_scope() as session:
679 last_active = session.execute(select(User.last_active).where(User.id == user.id)).scalar_one()
680 assert last_active > now() - timedelta(seconds=1)
682 # Check that last_active is untouched (since it was already updated recently)
683 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
684 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token),))
686 with session_scope() as session:
687 last_active_2 = session.execute(select(User.last_active).where(User.id == user.id)).scalar_one()
688 assert last_active_2 == last_active
690 # Check that activity is split by IP.
691 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
692 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token), ("x-couchers-real-ip", "1.1.1.1")))
694 with session_scope() as session:
695 api_calls = session.execute(
696 select(UserActivity.api_calls).where(UserActivity.ip_address == "1.1.1.1")
697 ).scalar_one()
698 assert api_calls == 1
700 # Check that activity is split in time bins.
701 # Update all UserActivity to be in the far past so that a new row is inserted on the next request.
702 with session_scope() as session:
703 session.execute(update(UserActivity).values(period=long_ago).where(UserActivity.user_id == user.id))
705 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
706 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token),))
708 with session_scope() as session:
709 api_calls = session.execute(
710 select(UserActivity.api_calls)
711 .where(UserActivity.user_id == user.id)
712 .order_by(UserActivity.id.desc())
713 .limit(1)
714 ).scalar_one()
715 assert api_calls == 1
718def test_tracing_interceptor_auth_cookies(db):
719 user, token = generate_user()
721 account = Account()
723 rpc_def = {
724 "rpc": account.GetAccountInfo,
725 "service_name": "org.couchers.api.account.Account",
726 "method_name": "GetAccountInfo",
727 "interceptors": [CouchersMiddlewareInterceptor()],
728 "request_type": empty_pb2.Empty,
729 "response_type": account_pb2.GetAccountInfoRes,
730 }
732 # with cookies
733 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
734 res1 = call_rpc(empty_pb2.Empty(), metadata=(("cookie", f"couchers-sesh={token}"),))
735 assert res1.username == user.username
737 with session_scope() as session:
738 trace = session.execute(select(APICall)).scalar_one()
739 assert trace.method == "/org.couchers.api.account.Account/GetAccountInfo"
740 assert not trace.status_code
741 assert trace.user_id == user.id
742 assert not trace.is_api_key
743 assert trace.request is not None
744 assert len(trace.request) == 0
745 assert not trace.traceback
748def test_tracing_interceptor_auth_api_key(db):
749 super_user, super_token = generate_user(is_superuser=True)
750 user, token = generate_user()
752 with real_admin_session(super_token) as api:
753 api.CreateApiKey(admin_pb2.CreateApiKeyReq(user=user.username))
755 with session_scope() as session:
756 api_key = session.execute(select(UserSession.token).where(UserSession.is_api_key)).scalar_one()
758 account = Account()
760 rpc_def = {
761 "rpc": account.GetAccountInfo,
762 "service_name": "org.couchers.api.account.Account",
763 "method_name": "GetAccountInfo",
764 "interceptors": [CouchersMiddlewareInterceptor()],
765 "request_type": empty_pb2.Empty,
766 "response_type": account_pb2.GetAccountInfoRes,
767 }
769 # with api key
770 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
771 res1 = call_rpc(empty_pb2.Empty(), metadata=(("authorization", f"Bearer {api_key}"),))
772 assert res1.username == user.username
774 with session_scope() as session:
775 trace = session.execute(
776 select(APICall).where(APICall.method == "/org.couchers.api.account.Account/GetAccountInfo")
777 ).scalar_one()
778 assert trace.method == "/org.couchers.api.account.Account/GetAccountInfo"
779 assert not trace.status_code
780 assert trace.user_id == user.id
781 assert trace.is_api_key
782 assert trace.request is not None
783 assert len(trace.request) == 0
784 assert not trace.traceback
787def test_auth_levels(db):
788 def TestRpc(request, context, session):
789 return empty_pb2.Empty()
791 def gen_args(service, method):
792 return {
793 "rpc": TestRpc,
794 "service_name": service,
795 "method_name": method,
796 "interceptors": [CouchersMiddlewareInterceptor()],
797 "request_type": empty_pb2.Empty,
798 "response_type": empty_pb2.Empty,
799 }
801 # superuser (note: superusers are automatically editors due to DB constraint)
802 _, super_token = generate_user(is_superuser=True)
803 # editor user
804 _, editor_token = generate_user(is_editor=True)
805 # normal user
806 _, normal_token = generate_user()
807 # jailed user
808 _, jailed_token = generate_user(accepted_tos=0)
809 # open user
810 open_token = ""
812 # pick some rpcs here with the right auth levels
813 open_args = gen_args("org.couchers.resources.Resources", "GetTermsOfService")
814 jailed_args = gen_args("org.couchers.jail.Jail", "JailInfo")
815 secure_args = gen_args("org.couchers.api.account.Account", "GetAccountInfo")
816 editor_args = gen_args("org.couchers.editor.Editor", "CreateCommunity")
817 admin_args = gen_args("org.couchers.admin.Admin", "GetUserDetails")
819 # pairs to check
820 checks = [
821 # name, args, token, works?, code, message
822 # open token only works on open servicers
823 ("open x open", open_token, open_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
824 ("open x jailed", open_token, jailed_args, False, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
825 ("open x secure", open_token, secure_args, False, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
826 ("open x editor", open_token, editor_args, False, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
827 ("open x admin", open_token, admin_args, False, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
828 # jailed works on jailed and open
829 ("jailed x open", jailed_token, open_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
830 ("jailed x jailed", jailed_token, jailed_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
831 ("jailed x secure", jailed_token, secure_args, False, grpc.StatusCode.UNAUTHENTICATED, "Permission denied"),
832 ("jailed x editor", jailed_token, editor_args, False, grpc.StatusCode.PERMISSION_DENIED, "Permission denied"),
833 ("jailed x admin", jailed_token, admin_args, False, grpc.StatusCode.PERMISSION_DENIED, "Permission denied"),
834 # normal works on all but editor and admin
835 ("normal x open", normal_token, open_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
836 ("normal x jailed", normal_token, jailed_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
837 ("normal x secure", normal_token, secure_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
838 ("normal x editor", normal_token, editor_args, False, grpc.StatusCode.PERMISSION_DENIED, "Permission denied"),
839 ("normal x admin", normal_token, admin_args, False, grpc.StatusCode.PERMISSION_DENIED, "Permission denied"),
840 # editor works on all but admin
841 ("editor x open", editor_token, open_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
842 ("editor x jailed", editor_token, jailed_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
843 ("editor x secure", editor_token, secure_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
844 ("editor x editor", editor_token, editor_args, True, grpc.StatusCode.PERMISSION_DENIED, "Permission denied"),
845 ("editor x admin", editor_token, admin_args, False, grpc.StatusCode.PERMISSION_DENIED, "Permission denied"),
846 # superuser works on all
847 ("super x open", super_token, open_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
848 ("super x jailed", super_token, jailed_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
849 ("super x secure", super_token, secure_args, True, grpc.StatusCode.UNAUTHENTICATED, "Unauthorized"),
850 ("super x editor", super_token, editor_args, True, grpc.StatusCode.PERMISSION_DENIED, "Permission denied"),
851 ("super x admin", super_token, admin_args, True, grpc.StatusCode.PERMISSION_DENIED, "Permission denied"),
852 ]
854 for name, token, args, should_work, code, message in checks:
855 print(f"Testing (token x args) = ({name}), {should_work=}")
856 metadata = (("cookie", f"couchers-sesh={token}"),)
857 with interceptor_dummy_api(**args) as call_rpc:
858 if should_work:
859 call_rpc(empty_pb2.Empty(), metadata=metadata)
860 else:
861 with pytest.raises(grpc.RpcError) as err:
862 call_rpc(empty_pb2.Empty(), metadata=metadata)
863 assert err.value.code() == code
864 assert err.value.details() == message
866 # a non-existent RPC
867 nonexistent = gen_args("org.couchers.nonexistent.NA", "GetNothing")
869 with interceptor_dummy_api(**nonexistent) as call_rpc:
870 with pytest.raises(grpc.RpcError) as err:
871 call_rpc(empty_pb2.Empty())
872 assert err.value.code() == grpc.StatusCode.UNIMPLEMENTED
873 assert err.value.details() == "API call does not exist. Please refresh and try again."
875 # an RPC without a service level
876 invalid_args = gen_args("org.couchers.media.Media", "UploadConfirmation")
878 with interceptor_dummy_api(**invalid_args) as call_rpc:
879 with pytest.raises(grpc.RpcError) as err:
880 call_rpc(empty_pb2.Empty())
881 assert err.value.code() == grpc.StatusCode.INTERNAL
882 assert err.value.details() == "Internal authentication error."
885def test_rejected_call_logged_unauthenticated(db):
886 method = "/org.couchers.api.account.Account/GetAccountInfo"
887 hist_before = _get_histogram_labels_value(method, "False", "", "UNAUTHENTICATED")
888 api_calls_before = _get_api_call_count(method, "unknown")
890 def TestRpc(request, context, session):
891 return empty_pb2.Empty()
893 with interceptor_dummy_api(
894 TestRpc,
895 interceptors=[CouchersMiddlewareInterceptor()],
896 service_name="org.couchers.api.account.Account",
897 method_name="GetAccountInfo",
898 ) as call_rpc:
899 with pytest.raises(grpc.RpcError) as e:
900 call_rpc(empty_pb2.Empty())
901 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED
903 with session_scope() as session:
904 trace = session.execute(select(APICall)).scalar_one()
905 assert trace.method == method
906 assert trace.status_code == "UNAUTHENTICATED"
907 assert trace.user_id is None
908 assert not trace.is_api_key
909 # the request is never deserialized on a reject path
910 assert trace.request is None
911 assert trace.response is None
912 assert not trace.traceback
913 assert trace.duration > 0
915 assert _get_histogram_labels_value(method, "False", "", "UNAUTHENTICATED") == hist_before + 1
916 assert _get_api_call_count(method, "unknown") == api_calls_before + 1
919def test_rejected_call_logged_permission_denied(db):
920 """A rejected call from a valid session is attributed to that user."""
921 user, token = generate_user()
922 method = "/org.couchers.admin.Admin/GetUserDetails"
923 hist_before = _get_histogram_labels_value(method, "True", "", "PERMISSION_DENIED")
925 def TestRpc(request, context, session):
926 return empty_pb2.Empty()
928 with interceptor_dummy_api(
929 TestRpc,
930 interceptors=[CouchersMiddlewareInterceptor()],
931 service_name="org.couchers.admin.Admin",
932 method_name="GetUserDetails",
933 ) as call_rpc:
934 with pytest.raises(grpc.RpcError) as e:
935 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token), ("x-couchers-client-platform", "web_mobile")))
936 assert e.value.code() == grpc.StatusCode.PERMISSION_DENIED
938 with session_scope() as session:
939 trace = session.execute(select(APICall)).scalar_one()
940 assert trace.method == method
941 assert trace.status_code == "PERMISSION_DENIED"
942 assert trace.user_id == user.id
943 assert trace.client_platform == ClientPlatform.web_mobile
944 assert not trace.traceback
946 # the same call is also counted in the per-user activity table, so the two agree
947 api_calls = session.execute(select(UserActivity.api_calls).where(UserActivity.user_id == user.id)).scalar_one()
948 assert api_calls == 1
950 assert _get_histogram_labels_value(method, "True", "", "PERMISSION_DENIED") == hist_before + 1
953def test_rejected_call_logged_service_missing_from_pool(db):
954 """A servicer registered for a service missing from the descriptor pool is our bug, so it keeps its name."""
955 method = "/org.couchers.nonexistent.NA/GetNothing"
956 hist_before = _get_histogram_labels_value(method, "False", "", "UNIMPLEMENTED")
957 api_calls_before = _get_api_call_count(method, "unknown")
958 bucketed_before = _get_histogram_labels_value(NONEXISTENT_METHOD_LABEL, "False", "", "UNIMPLEMENTED")
959 setup_db_before = _get_histogram_count(
960 servicer_setup_db_time_histogram, "couchers_servicer_setup_db_time_seconds_count", method=method
961 )
963 def TestRpc(request, context, session):
964 return empty_pb2.Empty()
966 with interceptor_dummy_api(
967 TestRpc,
968 interceptors=[CouchersMiddlewareInterceptor()],
969 service_name="org.couchers.nonexistent.NA",
970 method_name="GetNothing",
971 ) as call_rpc:
972 with pytest.raises(grpc.RpcError) as e:
973 call_rpc(empty_pb2.Empty(), metadata=(("x-couchers-real-ip", "1.1.1.1"),))
974 assert e.value.code() == grpc.StatusCode.UNIMPLEMENTED
975 assert e.value.details() == NONEXISTENT_API_CALL_ERROR_MESSAGE
977 with session_scope() as session:
978 trace = session.execute(select(APICall)).scalar_one()
979 assert trace.method == method
980 assert trace.status_code == "UNIMPLEMENTED"
981 assert trace.user_id is None
982 # headers are parsed before the auth level is looked up, so we know who made the call
983 assert trace.ip_address == "1.1.1.1"
984 assert trace.user_agent is not None
986 # the method is one this server registered, so it's a label of its own rather than bucketed
987 assert _get_histogram_labels_value(method, "False", "", "UNIMPLEMENTED") == hist_before + 1
988 assert _get_api_call_count(method, "unknown") == api_calls_before + 1
989 assert (
990 _get_histogram_count(
991 servicer_setup_db_time_histogram, "couchers_servicer_setup_db_time_seconds_count", method=method
992 )
993 == setup_db_before + 1
994 )
995 assert _get_histogram_labels_value(NONEXISTENT_METHOD_LABEL, "False", "", "UNIMPLEMENTED") == bucketed_before
998def test_rejected_call_logged_unregistered_method(db):
999 """A method with no servicer registered is terminated by the interceptor itself, and still logged."""
1000 hist_before = _get_histogram_labels_value(NONEXISTENT_METHOD_LABEL, "False", "", "UNIMPLEMENTED")
1002 def TestRpc(request, context, session):
1003 return empty_pb2.Empty()
1005 with interceptor_dummy_api(
1006 TestRpc,
1007 interceptors=[CouchersMiddlewareInterceptor()],
1008 call_method_name="NotRegistered",
1009 ) as call_rpc:
1010 with pytest.raises(grpc.RpcError) as e:
1011 call_rpc(empty_pb2.Empty(), metadata=(("x-couchers-real-ip", "1.1.1.1"),))
1012 assert e.value.code() == grpc.StatusCode.UNIMPLEMENTED
1013 assert e.value.details() == NONEXISTENT_API_CALL_ERROR_MESSAGE
1015 with session_scope() as session:
1016 trace = session.execute(select(APICall)).scalar_one()
1017 assert trace.method == "/org.couchers.auth.Auth/NotRegistered"
1018 assert trace.status_code == "UNIMPLEMENTED"
1019 assert trace.user_id is None
1020 assert trace.ip_address == "1.1.1.1"
1021 # no servicer was found, so the request was never deserialized
1022 assert trace.request is None
1024 assert _get_histogram_labels_value(NONEXISTENT_METHOD_LABEL, "False", "", "UNIMPLEMENTED") == hist_before + 1
1025 assert _get_histogram_labels_value("/org.couchers.auth.Auth/NotRegistered", "False", "", "UNIMPLEMENTED") == 0
1028def test_rejected_call_logged_missing_auth_level(db):
1029 def TestRpc(request, context, session):
1030 return empty_pb2.Empty()
1032 with interceptor_dummy_api(
1033 TestRpc,
1034 interceptors=[CouchersMiddlewareInterceptor()],
1035 service_name="org.couchers.media.Media",
1036 method_name="UploadConfirmation",
1037 ) as call_rpc:
1038 with pytest.raises(grpc.RpcError) as e:
1039 call_rpc(empty_pb2.Empty())
1040 assert e.value.code() == grpc.StatusCode.INTERNAL
1041 assert e.value.details() == MISSING_AUTH_LEVEL_ERROR_MESSAGE
1043 with session_scope() as session:
1044 trace = session.execute(select(APICall)).scalar_one()
1045 assert trace.method == "/org.couchers.media.Media/UploadConfirmation"
1046 assert trace.status_code == "INTERNAL"
1049def test_rejected_call_logged_bad_headers(db):
1050 _, token = generate_user()
1052 def TestRpc(request, context, session):
1053 return empty_pb2.Empty()
1055 with interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc:
1056 with pytest.raises(grpc.RpcError) as e:
1057 call_rpc(empty_pb2.Empty(), metadata=(cookie_auth(token), api_auth(token)))
1058 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED
1059 assert e.value.details() == COOKIES_AND_AUTH_HEADER_ERROR_MESSAGE
1061 with session_scope() as session:
1062 trace = session.execute(select(APICall)).scalar_one()
1063 assert trace.method == "/org.couchers.auth.Auth/SignupFlow"
1064 assert trace.status_code == "UNAUTHENTICATED"
1065 assert trace.user_id is None
1066 # the headers couldn't be parsed, so nothing is known about the client
1067 assert trace.ip_address is None
1068 assert trace.user_agent is None
1071def test_parse_headers_with_session_cookie():
1072 headers = {"cookie": "couchers-sesh=abc123; other-cookie=value"}
1073 result = parse_headers(headers)
1074 assert result.token == "abc123"
1075 assert result.is_api_key is False
1078def test_parse_headers_with_authorization_header():
1079 headers = {"authorization": "Bearer abc123"}
1080 result = parse_headers(headers)
1081 assert result.token == "abc123"
1082 assert result.is_api_key is True
1085def test_parse_headers_with_both_cookie_and_authorization():
1086 headers = {"cookie": "couchers-sesh=abc123", "authorization": "Bearer xyz789"}
1087 with pytest.raises(BadHeaders, match="Both cookies and authorization are present in headers"):
1088 parse_headers(headers)
1091def test_parse_headers_with_neither_cookie_nor_authorization():
1092 result = parse_headers({})
1093 assert result.token is None
1094 assert result.is_api_key is False
1097def test_parse_headers_with_all_optional_headers():
1098 headers = {
1099 "cookie": "couchers-sesh=abc123; couchers-user-id=42; NEXT_LOCALE=en",
1100 "x-couchers-real-ip": "192.168.1.1",
1101 "user-agent": "TestAgent/1.0",
1102 }
1103 result = parse_headers(headers)
1104 assert result.token == "abc123"
1105 assert result.is_api_key is False
1106 assert result.ip_address == "192.168.1.1"
1107 assert result.user_agent == "TestAgent/1.0"
1108 assert result.ui_lang == "en"
1109 assert result.user_id_str == "42"
1112def test_parse_headers_with_bytes_ip_address():
1113 headers: dict[str, str | bytes] = {
1114 "cookie": "couchers-sesh=abc123",
1115 "x-couchers-real-ip": b"192.168.1.1",
1116 }
1117 result = parse_headers(headers)
1118 assert result.ip_address is None
1121def test_parse_headers_with_bytes_user_agent():
1122 headers: dict[str, str | bytes] = {
1123 "cookie": "couchers-sesh=abc123",
1124 "user-agent": b"TestAgent/1.0",
1125 }
1126 result = parse_headers(headers)
1127 assert result.user_agent is None
1130def test_parse_headers_malformed_authorization():
1131 headers = {"authorization": "bearer abc123"}
1132 result = parse_headers(headers)
1133 assert result.token is None
1134 assert result.is_api_key is True
1137def test_auth_level_with_valid_service():
1138 result = get_proto_annotations().auth_level("/org.couchers.api.core.API/GetUser")
1139 assert result == annotations_pb2.AUTH_LEVEL_SECURE
1142def test_auth_level_with_nonexistent_service():
1143 with pytest.raises(CallRejectedError) as exc:
1144 get_proto_annotations().auth_level("/org.couchers.nonexistent.Service/Method")
1145 assert exc.value.msg == NONEXISTENT_API_CALL_ERROR_MESSAGE
1146 assert exc.value.code == grpc.StatusCode.UNIMPLEMENTED
1149def test_auth_level_with_unknown_auth_level():
1150 pool = Mock(spec=DescriptorPool)
1151 service_desc = Mock(spec=ServiceDescriptor)
1152 service_options = Mock()
1153 service_options.Extensions = {annotations_pb2.auth_level: annotations_pb2.AUTH_LEVEL_UNKNOWN}
1154 service_desc.GetOptions.return_value = service_options
1155 pool.FindServiceByName.return_value = service_desc
1157 with pytest.raises(CallRejectedError) as exc:
1158 ProtoAnnotations(pool).auth_level("/org.couchers.api.core.API/GetUser")
1159 assert exc.value.msg == MISSING_AUTH_LEVEL_ERROR_MESSAGE
1160 assert exc.value.code == grpc.StatusCode.INTERNAL
1163def test_validate_auth_level_with_unknown():
1164 with pytest.raises(CallRejectedError) as exc:
1165 validate_auth_level(annotations_pb2.AUTH_LEVEL_UNKNOWN)
1166 assert exc.value.msg == MISSING_AUTH_LEVEL_ERROR_MESSAGE
1167 assert exc.value.code == grpc.StatusCode.INTERNAL
1170def test_validate_auth_level_with_open():
1171 validate_auth_level(annotations_pb2.AUTH_LEVEL_OPEN)
1174def test_validate_auth_level_with_jailed():
1175 validate_auth_level(annotations_pb2.AUTH_LEVEL_JAILED)
1178def test_validate_auth_level_with_secure():
1179 validate_auth_level(annotations_pb2.AUTH_LEVEL_SECURE)
1182def test_validate_auth_level_with_editor():
1183 validate_auth_level(annotations_pb2.AUTH_LEVEL_EDITOR)
1186def test_validate_auth_level_with_admin():
1187 validate_auth_level(annotations_pb2.AUTH_LEVEL_ADMIN)
1190def test_check_auth_open_service_without_auth():
1191 check_permissions(None, annotations_pb2.AUTH_LEVEL_OPEN)
1194def test_check_auth_open_service_with_auth():
1195 auth_info = UserAuthInfo(
1196 user_id=1,
1197 is_jailed=False,
1198 is_editor=False,
1199 is_superuser=False,
1200 token_expiry=now(),
1201 ui_language_preference="en",
1202 timezone="Etc/UTC",
1203 token="abc123",
1204 is_api_key=False,
1205 )
1206 check_permissions(auth_info, annotations_pb2.AUTH_LEVEL_OPEN)
1209def test_check_auth_secure_service_without_auth():
1210 with pytest.raises(CallRejectedError):
1211 check_permissions(None, annotations_pb2.AUTH_LEVEL_SECURE)
1214def test_check_auth_secure_service_with_normal_auth():
1215 auth_info = UserAuthInfo(
1216 user_id=1,
1217 is_jailed=False,
1218 is_editor=False,
1219 is_superuser=False,
1220 token_expiry=now(),
1221 ui_language_preference="en",
1222 timezone="Etc/UTC",
1223 token="abc123",
1224 is_api_key=False,
1225 )
1226 check_permissions(auth_info, annotations_pb2.AUTH_LEVEL_SECURE)
1229def test_check_auth_secure_service_with_jailed_user():
1230 auth_info = UserAuthInfo(
1231 user_id=1,
1232 is_jailed=True,
1233 is_editor=False,
1234 is_superuser=False,
1235 token_expiry=now(),
1236 ui_language_preference="en",
1237 timezone="Etc/UTC",
1238 token="abc123",
1239 is_api_key=False,
1240 )
1241 with pytest.raises(CallRejectedError):
1242 check_permissions(auth_info, annotations_pb2.AUTH_LEVEL_SECURE)
1245def test_check_auth_jailed_service_with_jailed_user():
1246 auth_info = UserAuthInfo(
1247 user_id=1,
1248 is_jailed=True,
1249 is_editor=False,
1250 is_superuser=False,
1251 token_expiry=now(),
1252 ui_language_preference="en",
1253 timezone="Etc/UTC",
1254 token="abc123",
1255 is_api_key=False,
1256 )
1257 check_permissions(auth_info, annotations_pb2.AUTH_LEVEL_JAILED)
1260def test_check_auth_jailed_service_without_auth():
1261 with pytest.raises(CallRejectedError):
1262 check_permissions(None, annotations_pb2.AUTH_LEVEL_JAILED)
1265def test_check_auth_editor_service_without_editor():
1266 auth_info = UserAuthInfo(
1267 user_id=1,
1268 is_jailed=False,
1269 is_editor=False,
1270 is_superuser=False,
1271 token_expiry=now(),
1272 ui_language_preference="en",
1273 timezone="Etc/UTC",
1274 token="abc123",
1275 is_api_key=False,
1276 )
1277 with pytest.raises(CallRejectedError):
1278 check_permissions(auth_info, annotations_pb2.AUTH_LEVEL_EDITOR)
1281def test_check_auth_editor_service_with_editor():
1282 auth_info = UserAuthInfo(
1283 user_id=1,
1284 is_jailed=False,
1285 is_editor=True,
1286 is_superuser=False,
1287 token_expiry=now(),
1288 ui_language_preference="en",
1289 timezone="Etc/UTC",
1290 token="abc123",
1291 is_api_key=False,
1292 )
1293 check_permissions(auth_info, annotations_pb2.AUTH_LEVEL_EDITOR)
1296def test_check_auth_admin_service_without_superuser():
1297 auth_info = UserAuthInfo(
1298 user_id=1,
1299 is_jailed=False,
1300 is_editor=True,
1301 is_superuser=False,
1302 token_expiry=now(),
1303 ui_language_preference="en",
1304 timezone="Etc/UTC",
1305 token="abc123",
1306 is_api_key=False,
1307 )
1308 with pytest.raises(CallRejectedError):
1309 check_permissions(auth_info, annotations_pb2.AUTH_LEVEL_ADMIN)
1312def test_check_auth_admin_service_with_superuser():
1313 auth_info = UserAuthInfo(
1314 user_id=1,
1315 is_jailed=False,
1316 is_editor=True,
1317 is_superuser=True,
1318 token_expiry=now(),
1319 ui_language_preference="en",
1320 timezone="Etc/UTC",
1321 token="abc123",
1322 is_api_key=False,
1323 )
1324 check_permissions(auth_info, annotations_pb2.AUTH_LEVEL_ADMIN)
1327def test_check_auth_admin_service_without_auth():
1328 with pytest.raises(CallRejectedError):
1329 check_permissions(None, annotations_pb2.AUTH_LEVEL_ADMIN)
1332def test_parse_sofa_cookie_valid():
1333 sofa_value, cookie_string = generate_sofa_cookie()
1334 cookie_value = cookie_string.split("=", 1)[1].split(";")[0]
1336 headers = {"cookie": f"sofa={cookie_value}"}
1337 result = parse_sofa_cookie(headers)
1338 assert result == sofa_value
1341def test_parse_sofa_cookie_missing():
1342 headers = {"cookie": "other-cookie=value"}
1343 result = parse_sofa_cookie(headers)
1344 assert result is None
1347def test_parse_sofa_cookie_no_cookies():
1348 headers: dict[str, str] = {}
1349 result = parse_sofa_cookie(headers)
1350 assert result is None
1353def test_parse_sofa_cookie_invalid_base64():
1354 headers = {"cookie": "sofa=not-valid-base64!!!"}
1355 result = parse_sofa_cookie(headers)
1356 assert result is None
1359def test_parse_sofa_cookie_invalid_encryption():
1360 headers = {"cookie": f"sofa={b64encode(b'invalid encrypted data')}"}
1361 result = parse_sofa_cookie(headers)
1362 assert result is None
1365def test_parse_sofa_cookie_invalid_proto():
1366 encrypted = simple_encrypt("sofa_cookie", b"not a valid proto")
1367 headers = {"cookie": f"sofa={b64encode(encrypted)}"}
1368 result = parse_sofa_cookie(headers)
1369 assert result is not None or result is None
1372def test_generate_sofa_cookie():
1373 sofa_value, cookie_string = generate_sofa_cookie()
1375 assert sofa_value
1376 assert isinstance(sofa_value, str)
1377 assert len(sofa_value) > 20
1379 assert "sofa=" in cookie_string
1380 assert "expires=" in cookie_string.lower()
1382 cookie_value = cookie_string.split("=", 1)[1].split(";")[0]
1383 headers = {"cookie": f"sofa={cookie_value}"}
1384 parsed_value = parse_sofa_cookie(headers)
1385 assert parsed_value == sofa_value
1388def test_parse_headers_with_sofa_cookie():
1389 sofa_value, cookie_string = generate_sofa_cookie()
1390 cookie_value = cookie_string.split("=", 1)[1].split(";")[0]
1392 headers = {
1393 "cookie": f"couchers-sesh=abc123; sofa={cookie_value}",
1394 }
1395 result = parse_headers(headers)
1396 assert result.token == "abc123"
1397 assert result.sofa == sofa_value
1400def test_parse_headers_without_sofa_cookie():
1401 headers = {
1402 "cookie": "couchers-sesh=abc123",
1403 }
1404 result = parse_headers(headers)
1405 assert result.token == "abc123"
1406 assert result.sofa is None
1409def test_sofa_cookie_logged_new(db):
1410 def TestRpc(request, context, session):
1411 return empty_pb2.Empty()
1413 with interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc:
1414 call_rpc(empty_pb2.Empty())
1416 with session_scope() as session:
1417 trace = session.execute(select(APICall)).scalar_one()
1418 assert trace.sofa is not None
1419 assert len(trace.sofa) > 20
1422def test_sofa_cookie_logged_existing(db):
1423 sofa_value, cookie_string = generate_sofa_cookie()
1424 cookie_value = cookie_string.split("=", 1)[1].split(";")[0]
1426 def TestRpc(request, context, session):
1427 return empty_pb2.Empty()
1429 with interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc:
1430 call_rpc(empty_pb2.Empty(), metadata=(("cookie", f"sofa={cookie_value}"),))
1432 with session_scope() as session:
1433 trace = session.execute(select(APICall)).scalar_one()
1434 assert trace.sofa == sofa_value
1437def test_sofa_cookie_logged_invalid_generates_new(db):
1438 def TestRpc(request, context, session):
1439 return empty_pb2.Empty()
1441 with interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc:
1442 call_rpc(empty_pb2.Empty(), metadata=(("cookie", "sofa=invalid-cookie-value"),))
1444 with session_scope() as session:
1445 trace = session.execute(select(APICall)).scalar_one()
1446 assert trace.sofa is not None
1447 assert trace.sofa != "invalid-cookie-value"
1448 assert len(trace.sofa) > 20
1451def test_sofa_cookie_with_authenticated_user(db):
1452 user, token = generate_user()
1453 sofa_value, cookie_string = generate_sofa_cookie()
1454 cookie_value = cookie_string.split("=", 1)[1].split(";")[0]
1456 account = Account()
1458 rpc_def = {
1459 "rpc": account.GetAccountInfo,
1460 "service_name": "org.couchers.api.account.Account",
1461 "method_name": "GetAccountInfo",
1462 "interceptors": [CouchersMiddlewareInterceptor()],
1463 "request_type": empty_pb2.Empty,
1464 "response_type": account_pb2.GetAccountInfoRes,
1465 }
1467 with interceptor_dummy_api(**rpc_def, creds=grpc.local_channel_credentials()) as call_rpc:
1468 res = call_rpc(empty_pb2.Empty(), metadata=(("cookie", f"couchers-sesh={token}; sofa={cookie_value}"),))
1469 assert res.username == user.username
1471 with session_scope() as session:
1472 trace = session.execute(select(APICall)).scalar_one()
1473 assert trace.user_id == user.id
1474 assert trace.sofa == sofa_value
1477def test_sofa_cookie_persists_on_exception(db):
1478 sofa_value, cookie_string = generate_sofa_cookie()
1479 cookie_value = cookie_string.split("=", 1)[1].split(";")[0]
1481 def TestRpc(request, context, session):
1482 raise Exception("Test error")
1484 with interceptor_dummy_api(TestRpc, interceptors=[CouchersMiddlewareInterceptor()]) as call_rpc:
1485 with pytest.raises(Exception, match="Test error"):
1486 call_rpc(empty_pb2.Empty(), metadata=(("cookie", f"sofa={cookie_value}"),))
1488 with session_scope() as session:
1489 trace = session.execute(select(APICall)).scalar_one()
1490 assert trace.sofa == sofa_value
1491 assert trace.traceback is not None
1492 assert "Test error" in trace.traceback