Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/

1import logging

2from abc import ABC, abstractmethod

3from asyncio import IncompleteReadError, StreamReader

4from typing import Awaitable, Callable, List, Optional, Protocol, Union

6from redis.maint_notifications import (

7 MaintenanceNotification,

8 NodeFailedOverNotification,

9 NodeFailingOverNotification,

10 NodeMigratedNotification,

11 NodeMigratingNotification,

12 NodeMovingNotification,

13 OSSNodeMigratedNotification,

14 OSSNodeMigratingNotification,

15)

16from redis.utils import deprecated_function, safe_str

18from ..exceptions import (

19 AskError,

20 AuthenticationError,

21 AuthenticationWrongNumberOfArgsError,

22 BusyLoadingError,

23 ClusterCrossSlotError,

24 ClusterDownError,

25 ConnectionError,

26 ExecAbortError,

27 ExternalAuthProviderError,

28 InvalidResponse,

29 MasterDownError,

30 ModuleError,

31 MovedError,

32 NoPermissionError,

33 NoScriptError,

34 NoSuchFieldsetError,

35 OutOfMemoryError,

36 ReadOnlyError,

37 ResponseError,

38 TryAgainError,

39)

40from ..typing import EncodableT

41from .encoders import Encoder

42from .socket import SERVER_CLOSED_CONNECTION_ERROR, SocketBuffer

44# Errors a parser can raise from the middle of a reply, after the read cursor

45# has already advanced past bytes the parser cannot re-interpret. The resume

46# machinery from #2510/#2695 rewinds so an *interrupted* read can be re-parsed

47# from the start; it cannot help here, because re-parsing the same bytes just

48# reproduces the same failure. Connection.read_response drops the connection on

49# these regardless of disconnect_on_error, or the caller loops forever (#4291).

50#

51# - InvalidResponse: unknown reply type byte, or hiredis' own nesting limit.

52# - UnicodeDecodeError: Encoder.decode at the tail of _read_response, reached

53# only after the payload has been consumed (decode_responses=True subscriber

54# receiving a binary PUBLISH payload).

55# - RecursionError: deeply nested aggregate replies exhaust the stack in the

56# pure-Python parsers. #4144 turns this into InvalidResponse at a bounded

57# depth; until then, catching it here is what stops the loop.

58#

59# Malformed numeric frames (`:abc\r\n`, `$xyz\r\n`, `*abc\r\n`) are converted to

60# InvalidResponse by the pure-Python parsers (_parse_int/_parse_float below), so

61# they land here too. Plain ValueError is deliberately NOT in this tuple: a

62# user push handler may raise it after the frame was fully consumed, and that

63# must not tear down a healthy connection.

64UNRECOVERABLE_PARSE_ERRORS = (InvalidResponse, UnicodeDecodeError, RecursionError)

67def _parse_int(value: bytes, raw: bytes) -> int:

68 """int() for a RESP numeric field; a non-numeric field is a protocol error."""

69 try:

70 return int(value)

71 except ValueError:

72 raise InvalidResponse(f"Protocol Error: {raw!r}") from None

75def _parse_float(value: bytes, raw: bytes) -> float:

76 try:

77 return float(value)

78 except ValueError:

79 raise InvalidResponse(f"Protocol Error: {raw!r}") from None

82MODULE_LOAD_ERROR = "Error loading the extension. Please check the server logs."

83NO_SUCH_MODULE_ERROR = "Error unloading module: no such module with that name"

84MODULE_UNLOAD_NOT_POSSIBLE_ERROR = "Error unloading module: operation not possible."

85MODULE_EXPORTS_DATA_TYPES_ERROR = (

86 "Error unloading module: the module "

87 "exports one or more module-side data "

88 "types, can't unload"

89)

90# user send an AUTH cmd to a server without authorization configured

91NO_AUTH_SET_ERROR = {

92 # Redis >= 6.0

93 "AUTH <password> called without any password "

94 "configured for the default user. Are you sure "

95 "your configuration is correct?": AuthenticationError,

96 # Redis < 6.0

97 "Client sent AUTH, but no password is set": AuthenticationError,

98}

100EXTERNAL_AUTH_PROVIDER_ERROR = {

101 "problem with LDAP service": ExternalAuthProviderError,

102}

103

104# HIMPORT SET referencing a fieldset the connection has not prepared. The server

105# reply is a fixed message with no fieldset name appended (verified against the

106# server: always exactly ``ERR no such fieldset``), so an exact match is correct.

107NO_SUCH_FIELDSET_ERROR = {

108 "no such fieldset": NoSuchFieldsetError,

109}

110

111logger = logging.getLogger(__name__)

112

113

114class BaseParser(ABC):

115 EXCEPTION_CLASSES = {

116 "ERR": {

117 "max number of clients reached": ConnectionError,

118 "invalid password": AuthenticationError,

119 # some Redis server versions report invalid command syntax

120 # in lowercase

121 "wrong number of arguments "

122 "for 'auth' command": AuthenticationWrongNumberOfArgsError,

123 # some Redis server versions report invalid command syntax

124 # in uppercase

125 "wrong number of arguments "

126 "for 'AUTH' command": AuthenticationWrongNumberOfArgsError,

127 MODULE_LOAD_ERROR: ModuleError,

128 MODULE_EXPORTS_DATA_TYPES_ERROR: ModuleError,

129 NO_SUCH_MODULE_ERROR: ModuleError,

130 MODULE_UNLOAD_NOT_POSSIBLE_ERROR: ModuleError,

131 **NO_AUTH_SET_ERROR,

132 **EXTERNAL_AUTH_PROVIDER_ERROR,

133 **NO_SUCH_FIELDSET_ERROR,

134 },

135 "OOM": OutOfMemoryError,

136 "WRONGPASS": AuthenticationError,

137 "EXECABORT": ExecAbortError,

138 "LOADING": BusyLoadingError,

139 "NOSCRIPT": NoScriptError,

140 "READONLY": ReadOnlyError,

141 "NOAUTH": AuthenticationError,

142 "NOPERM": NoPermissionError,

143 "ASK": AskError,

144 "TRYAGAIN": TryAgainError,

145 "MOVED": MovedError,

146 "CLUSTERDOWN": ClusterDownError,

147 "CROSSSLOT": ClusterCrossSlotError,

148 "MASTERDOWN": MasterDownError,

149 }

150

151 @classmethod

152 def parse_error(cls, response):

153 "Parse an error response"

154 error_code = response.split(" ")[0]

155 if error_code in cls.EXCEPTION_CLASSES:

156 response = response[len(error_code) + 1 :]

157 exception_class = cls.EXCEPTION_CLASSES[error_code]

158 if isinstance(exception_class, dict):

159 exception_class = exception_class.get(response, ResponseError)

160 return exception_class(response, status_code=error_code)

161 return ResponseError(response)

162

163 @abstractmethod

164 def on_disconnect(self):

165 pass

166

167 @abstractmethod

168 def on_connect(self, connection):

169 pass

170

171

172class _RESPBase(BaseParser):

173 """Base class for sync-based resp parsing"""

174

175 def __init__(self, socket_read_size):

176 self.socket_read_size = socket_read_size

177 self.encoder = None

178 self._sock = None

179 self._buffer = None

180

181 def __del__(self):

182 try:

183 self.on_disconnect()

184 except Exception:

185 pass

186

187 def on_connect(self, connection):

188 "Called when the socket connects"

189 self._sock = connection._sock

190 self._buffer = SocketBuffer(

191 self._sock, self.socket_read_size, connection.socket_timeout

192 )

193 self.encoder = connection.encoder

194

195 def on_disconnect(self):

196 "Called when the socket disconnects"

197 self._sock = None

198 if self._buffer is not None:

199 self._buffer.close()

200 self._buffer = None

201 self.encoder = None

202

203 def can_read(self, timeout: float = 0) -> bool:

204 # TODO: Rename this API; it detects pending data or dirty/closed

205 # connection state, not only whether application data can be read.

206 if self._buffer is None:

207 return False

208 return self._buffer.can_read(timeout)

209

210

211class AsyncBaseParser(BaseParser):

212 """Base parsing class for the python-backed async parser"""

213

214 __slots__ = "_stream", "_read_size", "_connected"

215

216 def __init__(self, socket_read_size: int):

217 self._stream: Optional[StreamReader] = None

218 self._read_size = socket_read_size

219 # Initialized here, not only in on_connect() / on_disconnect(): the

220 # read guards in can_read() and the subclasses' read_response() consult

221 # this flag to turn a read on a dead connection into a retryable

222 # ConnectionError. A parser that has never connected would otherwise

223 # raise AttributeError from those guards, which no retry layer acts on.

224 self._connected: bool = False

225

226 @deprecated_function(

227 version="8.0.0", reason="Use can_read() instead", name="can_read_destructive"

228 )

229 @abstractmethod

230 async def can_read_destructive(self) -> bool:

231 pass

232

233 @abstractmethod

234 async def can_read(self) -> bool:

235 # TODO: Rename this API; it detects pending data or dirty/closed

236 # connection state, not only whether application data can be read.

237 pass

238

239 async def read_response(

240 self, disable_decoding: bool = False

241 ) -> Union[EncodableT, ResponseError, None, List[EncodableT]]:

242 raise NotImplementedError()

243

244

245class MaintenanceNotificationsParser:

246 """Protocol defining maintenance push notification parsing functionality"""

247

248 @staticmethod

249 def parse_oss_maintenance_start_msg(response):

250 # Expected message format is:

251 # SMIGRATING <seq_number> <slot, range1-range2,...>

252 id = response[1]

253 slots = safe_str(response[2])

254 return OSSNodeMigratingNotification(id, slots)

255

256 @staticmethod

257 def parse_oss_maintenance_completed_msg(response):

258 # Expected message format is:

259 # SMIGRATED <seq_number> [[<src_host:port> <dest_host:port> <slot_range>], ...]

260 id = response[1]

261 nodes_to_slots_mapping_data = response[2]

262 # Build the nodes_to_slots_mapping dict structure:

263 # {

264 # "src_host:port": [

265 # {"dest_host:port": "slot_range"},

266 # ...

267 # ],

268 # ...

269 # }

270 nodes_to_slots_mapping = {}

271 for src_node, dest_node, slots in nodes_to_slots_mapping_data:

272 src_node_str = safe_str(src_node)

273 dest_node_str = safe_str(dest_node)

274 slots_str = safe_str(slots)

275

276 if src_node_str not in nodes_to_slots_mapping:

277 nodes_to_slots_mapping[src_node_str] = []

278 nodes_to_slots_mapping[src_node_str].append({dest_node_str: slots_str})

279

280 return OSSNodeMigratedNotification(id, nodes_to_slots_mapping)

281

282 @staticmethod

283 def parse_maintenance_start_msg(response, notification_type):

284 # Expected message format is: <notification_type> <seq_number> <time>

285 # Examples:

286 # MIGRATING 1 10

287 # FAILING_OVER 2 20

288 id = response[1]

289 ttl = response[2]

290 return notification_type(id, ttl)

291

292 @staticmethod

293 def parse_maintenance_completed_msg(response, notification_type):

294 # Expected message format is: <notification_type> <seq_number>

295 # Examples:

296 # MIGRATED 1

297 # FAILED_OVER 2

298 id = response[1]

299 return notification_type(id)

300

301 @staticmethod

302 def parse_moving_msg(response):

303 # Expected message format is: MOVING <seq_number> <time> <endpoint>

304 id = response[1]

305 ttl = response[2]

306 if response[3] is None:

307 host, port = None, None

308 else:

309 value = safe_str(response[3])

310 host, port = value.rsplit(":", 1)

311 port = int(port) if port is not None else None

312

313 return NodeMovingNotification(id, host, port, ttl)

314

315

316_INVALIDATION_MESSAGE = "invalidate"

317_MOVING_MESSAGE = "MOVING"

318_MIGRATING_MESSAGE = "MIGRATING"

319_MIGRATED_MESSAGE = "MIGRATED"

320_FAILING_OVER_MESSAGE = "FAILING_OVER"

321_FAILED_OVER_MESSAGE = "FAILED_OVER"

322_SMIGRATING_MESSAGE = "SMIGRATING"

323_SMIGRATED_MESSAGE = "SMIGRATED"

324

325_MAINTENANCE_MESSAGES = (

326 _MIGRATING_MESSAGE,

327 _MIGRATED_MESSAGE,

328 _FAILING_OVER_MESSAGE,

329 _FAILED_OVER_MESSAGE,

330 _SMIGRATING_MESSAGE,

331)

332

333MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING: dict[

334 str, tuple[type[MaintenanceNotification], Callable]

335] = {

336 _MIGRATING_MESSAGE: (

337 NodeMigratingNotification,

338 MaintenanceNotificationsParser.parse_maintenance_start_msg,

339 ),

340 _MIGRATED_MESSAGE: (

341 NodeMigratedNotification,

342 MaintenanceNotificationsParser.parse_maintenance_completed_msg,

343 ),

344 _FAILING_OVER_MESSAGE: (

345 NodeFailingOverNotification,

346 MaintenanceNotificationsParser.parse_maintenance_start_msg,

347 ),

348 _FAILED_OVER_MESSAGE: (

349 NodeFailedOverNotification,

350 MaintenanceNotificationsParser.parse_maintenance_completed_msg,

351 ),

352 _MOVING_MESSAGE: (

353 NodeMovingNotification,

354 MaintenanceNotificationsParser.parse_moving_msg,

355 ),

356 _SMIGRATING_MESSAGE: (

357 OSSNodeMigratingNotification,

358 MaintenanceNotificationsParser.parse_oss_maintenance_start_msg,

359 ),

360 _SMIGRATED_MESSAGE: (

361 OSSNodeMigratedNotification,

362 MaintenanceNotificationsParser.parse_oss_maintenance_completed_msg,

363 ),

364}

365

366

367class PushNotificationsParser(Protocol):

368 """Protocol defining RESP3-specific parsing functionality"""

369

370 pubsub_push_handler_func: Callable

371 invalidation_push_handler_func: Optional[Callable] = None

372 node_moving_push_handler_func: Optional[Callable] = None

373 maintenance_push_handler_func: Optional[Callable] = None

374 oss_cluster_maint_push_handler_func: Optional[Callable] = None

375

376 def handle_pubsub_push_response(self, response):

377 """Handle pubsub push responses"""

378 raise NotImplementedError()

379

380 def handle_push_response(self, response, **kwargs):

381 msg_type = response[0]

382 if isinstance(msg_type, bytes):

383 msg_type = msg_type.decode()

384

385 if msg_type not in (

386 _INVALIDATION_MESSAGE,

387 *_MAINTENANCE_MESSAGES,

388 _MOVING_MESSAGE,

389 _SMIGRATED_MESSAGE,

390 ):

391 return self.pubsub_push_handler_func(response)

392

393 try:

394 if (

395 msg_type == _INVALIDATION_MESSAGE

396 and self.invalidation_push_handler_func

397 ):

398 return self.invalidation_push_handler_func(response)

399

400 if msg_type == _MOVING_MESSAGE and self.node_moving_push_handler_func:

401 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[

402 msg_type

403 ][1]

404

405 notification = parser_function(response)

406 return self.node_moving_push_handler_func(notification)

407

408 if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func:

409 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[

410 msg_type

411 ][1]

412 if msg_type == _SMIGRATING_MESSAGE:

413 notification = parser_function(response)

414 else:

415 notification_type = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[

416 msg_type

417 ][0]

418 notification = parser_function(response, notification_type)

419

420 if notification is not None:

421 return self.maintenance_push_handler_func(notification)

422 if msg_type == _SMIGRATED_MESSAGE and (

423 self.oss_cluster_maint_push_handler_func

424 or self.maintenance_push_handler_func

425 ):

426 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[

427 msg_type

428 ][1]

429 notification = parser_function(response)

430

431 if notification is not None:

432 if self.maintenance_push_handler_func:

433 self.maintenance_push_handler_func(notification)

434 if self.oss_cluster_maint_push_handler_func:

435 self.oss_cluster_maint_push_handler_func(notification)

436 except Exception as e:

437 logger.error(

438 "Error handling {} message ({}): {}".format(msg_type, response, e)

439 )

440

441 return None

442

443 def set_pubsub_push_handler(self, pubsub_push_handler_func):

444 self.pubsub_push_handler_func = pubsub_push_handler_func

445

446 def set_invalidation_push_handler(self, invalidation_push_handler_func):

447 self.invalidation_push_handler_func = invalidation_push_handler_func

448

449 def set_node_moving_push_handler(self, node_moving_push_handler_func):

450 self.node_moving_push_handler_func = node_moving_push_handler_func

451

452 def set_maintenance_push_handler(self, maintenance_push_handler_func):

453 self.maintenance_push_handler_func = maintenance_push_handler_func

454

455 def set_oss_cluster_maint_push_handler(self, oss_cluster_maint_push_handler_func):

456 self.oss_cluster_maint_push_handler_func = oss_cluster_maint_push_handler_func

457

458

459class AsyncPushNotificationsParser(Protocol):

460 """Protocol defining async RESP3-specific parsing functionality"""

461

462 pubsub_push_handler_func: Callable

463 invalidation_push_handler_func: Optional[Callable] = None

464 node_moving_push_handler_func: Optional[Callable[..., Awaitable[None]]] = None

465 maintenance_push_handler_func: Optional[Callable[..., Awaitable[None]]] = None

466 oss_cluster_maint_push_handler_func: Optional[Callable[..., Awaitable[None]]] = None

467

468 async def handle_pubsub_push_response(self, response):

469 """Handle pubsub push responses asynchronously"""

470 raise NotImplementedError()

471

472 async def handle_push_response(self, response, **kwargs):

473 """Handle push responses asynchronously"""

474

475 msg_type = response[0]

476 if isinstance(msg_type, bytes):

477 msg_type = msg_type.decode()

478

479 if msg_type not in (

480 _INVALIDATION_MESSAGE,

481 *_MAINTENANCE_MESSAGES,

482 _MOVING_MESSAGE,

483 _SMIGRATED_MESSAGE,

484 ):

485 return await self.pubsub_push_handler_func(response)

486

487 try:

488 if (

489 msg_type == _INVALIDATION_MESSAGE

490 and self.invalidation_push_handler_func

491 ):

492 return await self.invalidation_push_handler_func(response)

493

494 if isinstance(msg_type, bytes):

495 msg_type = msg_type.decode()

496

497 if msg_type == _MOVING_MESSAGE and self.node_moving_push_handler_func:

498 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[

499 msg_type

500 ][1]

501 notification = parser_function(response)

502 return await self.node_moving_push_handler_func(notification)

503

504 if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func:

505 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[

506 msg_type

507 ][1]

508 if msg_type == _SMIGRATING_MESSAGE:

509 notification = parser_function(response)

510 else:

511 notification_type = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[

512 msg_type

513 ][0]

514 notification = parser_function(response, notification_type)

515

516 if notification is not None:

517 return await self.maintenance_push_handler_func(notification)

518 if (

519 msg_type == _SMIGRATED_MESSAGE

520 and self.oss_cluster_maint_push_handler_func

521 ):

522 parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[

523 msg_type

524 ][1]

525 notification = parser_function(response)

526 if notification is not None:

527 return await self.oss_cluster_maint_push_handler_func(notification)

528 except Exception as e:

529 logger.error(

530 "Error handling {} message ({}): {}".format(msg_type, response, e)

531 )

532

533 return None

534

535 def set_pubsub_push_handler(self, pubsub_push_handler_func):

536 """Set the pubsub push handler function"""

537 self.pubsub_push_handler_func = pubsub_push_handler_func

538

539 def set_invalidation_push_handler(self, invalidation_push_handler_func):

540 """Set the invalidation push handler function"""

541 self.invalidation_push_handler_func = invalidation_push_handler_func

542

543 def set_node_moving_push_handler(self, node_moving_push_handler_func):

544 self.node_moving_push_handler_func = node_moving_push_handler_func

545

546 def set_maintenance_push_handler(self, maintenance_push_handler_func):

547 self.maintenance_push_handler_func = maintenance_push_handler_func

548

549 def set_oss_cluster_maint_push_handler(self, oss_cluster_maint_push_handler_func):

550 self.oss_cluster_maint_push_handler_func = oss_cluster_maint_push_handler_func

551

552

553class _AsyncRESPBase(AsyncBaseParser):

554 """Base class for async resp parsing"""

555

556 __slots__ = AsyncBaseParser.__slots__ + ("encoder", "_buffer", "_pos", "_chunks")

557

558 def __init__(self, socket_read_size: int):

559 super().__init__(socket_read_size)

560 self.encoder: Optional[Encoder] = None

561 self._buffer = b""

562 self._chunks = []

563 self._pos = 0

564

565 def _clear(self):

566 self._buffer = b""

567 self._chunks.clear()

568

569 def on_connect(self, connection):

570 """Called when the stream connects"""

571 self._stream = connection._reader

572 if self._stream is None:

573 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)

574 self.encoder = connection.encoder

575 self._clear()

576 self._connected = True

577

578 def on_disconnect(self):

579 """Called when the stream disconnects"""

580 self._connected = False

581

582 @deprecated_function(

583 version="8.0.0",

584 reason="Use can_read() instead",

585 name="can_read_destructive",

586 )

587 async def can_read_destructive(self) -> bool:

588 return await self.can_read()

589

590 async def can_read(self) -> bool:

591 # TODO: Rename this API; it detects pending data or dirty/closed

592 # connection state, not only whether application data can be read.

593 if not self._connected:

594 raise OSError("Buffer is closed.")

595 # buffered data wins over EOF, like the sync SocketBuffer: a pending

596 # response or push notification must stay readable even if the server

597 # has since closed the connection.

598 if self._buffer:

599 return True

600 if self._stream.at_eof():

601 # Raise like the sync SocketBuffer does on a server-closed

602 # connection, so callers that tolerate pending data (push

603 # notifications) can't mistake EOF for a readable connection.

604 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)

605 # asyncio.StreamReader has no public non-destructive API for checking

606 # buffered bytes. Preserve dirty-connection detection for the Python

607 # parser and fail loudly if the private buffer API changes.

608 return bool(self._stream._buffer)

609

610 async def _read(self, length: int) -> bytes:

611 """

612 Read `length` bytes of data. These are assumed to be followed

613 by a '\r\n' terminator which is subsequently discarded.

614 """

615 want = length + 2

616 end = self._pos + want

617 if len(self._buffer) >= end:

618 result = self._buffer[self._pos : end - 2]

619 else:

620 tail = self._buffer[self._pos :]

621 try:

622 data = await self._stream.readexactly(want - len(tail))

623 except IncompleteReadError as error:

624 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from error

625 result = (tail + data)[:-2]

626 self._chunks.append(data)

627 self._pos += want

628 return result

629

630 async def _readline(self) -> bytes:

631 """

632 read an unknown number of bytes up to the next '\r\n'

633 line separator, which is discarded.

634 """

635 found = self._buffer.find(b"\r\n", self._pos)

636 if found >= 0:

637 result = self._buffer[self._pos : found]

638 else:

639 tail = self._buffer[self._pos :]

640 data = await self._stream.readline()

641 if not data.endswith(b"\r\n"):

642 raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)

643 result = (tail + data)[:-2]

644 self._chunks.append(data)

645 self._pos += len(result) + 2

646 return result

Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/_parsers/base.py: 35%

291 statements