(
client_greeting: bytes,
server_choice: bytes,
client_auth: bytes,
server_resp: bytes,
address: bytes,
packed: bytes,
tctx: Context,
)
| 488 | ], |
| 489 | ) |
| 490 | def test_socks5_auth_success( |
| 491 | client_greeting: bytes, |
| 492 | server_choice: bytes, |
| 493 | client_auth: bytes, |
| 494 | server_resp: bytes, |
| 495 | address: bytes, |
| 496 | packed: bytes, |
| 497 | tctx: Context, |
| 498 | ): |
| 499 | ProxyAuth().load(tctx.options) |
| 500 | tctx.options.proxyauth = "user:password" |
| 501 | server = Placeholder(Server) |
| 502 | nextlayer = Placeholder(NextLayer) |
| 503 | playbook = ( |
| 504 | Playbook(modes.Socks5Proxy(tctx), logs=True) |
| 505 | >> DataReceived(tctx.client, client_greeting) |
| 506 | << SendData(tctx.client, server_choice) |
| 507 | >> DataReceived(tctx.client, client_auth) |
| 508 | << modes.Socks5AuthHook(Placeholder(modes.Socks5AuthData)) |
| 509 | >> reply(side_effect=_valid_socks_auth) |
| 510 | << SendData(tctx.client, server_resp) |
| 511 | >> DataReceived( |
| 512 | tctx.client, b"\x05\x01\x00" + packed + b"\x12\x34applicationdata" |
| 513 | ) |
| 514 | << OpenConnection(server) |
| 515 | >> reply(None) |
| 516 | << SendData(tctx.client, b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00") |
| 517 | << NextLayerHook(nextlayer) |
| 518 | ) |
| 519 | assert playbook |
| 520 | assert server().address == (address, 0x1234) |
| 521 | assert nextlayer().data_client() == b"applicationdata" |
| 522 | |
| 523 | |
| 524 | @pytest.mark.parametrize( |
nothing calls this directly
no test coverage detected
searching dependent graphs…