from unittest import mock import codecs import pytest import hyperframe from mitmproxy.net import tcp, http from mitmproxy.net.http import http2 from mitmproxy import exceptions from ...mitmproxy.net import tservers as net_tservers from pathod.protocols.http2 import HTTP2StateProtocol, TCPHandler class TestTCPHandlerWrapper: def test_wrapped(self): h = TCPHandler(rfile='foo', wfile='bar') p = HTTP2StateProtocol(h) assert p.tcp_handler.rfile == 'foo' assert p.tcp_handler.wfile == 'bar' def test_direct(self): p = HTTP2StateProtocol(rfile='foo', wfile='bar') assert isinstance(p.tcp_handler, TCPHandler) assert p.tcp_handler.rfile == 'foo' assert p.tcp_handler.wfile == 'bar' class EchoHandler(tcp.BaseHandler): sni = None def handle(self): while True: v = self.rfile.safe_read(1) self.wfile.write(v) self.wfile.flush() class TestProtocol: @mock.patch("pathod.protocols.http2.HTTP2StateProtocol.perform_server_connection_preface") @mock.patch("pathod.protocols.http2.HTTP2StateProtocol.perform_client_connection_preface") def test_perform_connection_preface(self, mock_client_method, mock_server_method): protocol = HTTP2StateProtocol(is_server=False) protocol.connection_preface_performed = True protocol.perform_connection_preface() assert not mock_client_method.called assert not mock_server_method.called protocol.perform_connection_preface(force=True) assert mock_client_method.called assert not mock_server_method.called @mock.patch("pathod.protocols.http2.HTTP2StateProtocol.perform_server_connection_preface") @mock.patch("pathod.protocols.http2.HTTP2StateProtocol.perform_client_connection_preface") def test_perform_connection_preface_server(self, mock_client_method, mock_server_method): protocol = HTTP2StateProtocol(is_server=True) protocol.connection_preface_performed = True protocol.perform_connection_preface() assert not mock_client_method.called assert not mock_server_method.called protocol.perform_connection_preface(force=True) assert not mock_client_method.called assert mock_server_method.called class TestCheckALPNMatch(net_tservers.ServerTestBase): handler = EchoHandler ssl = dict( alpn_select=b'h2', ) def test_check_alpn(self): c = tcp.TCPClient(("127.0.0.1", self.port)) with c.connect(): c.convert_to_ssl(alpn_protos=[b'h2']) protocol = HTTP2StateProtocol(c) assert protocol.check_alpn() class TestCheckALPNMismatch(net_tservers.ServerTestBase): handler = EchoHandler ssl = dict( alpn_select=None, ) def test_check_alpn(self): c = tcp.TCPClient(("127.0.0.1", self.port)) with c.connect(): c.convert_to_ssl(alpn_protos=[b'h2']) protocol = HTTP2StateProtocol(c) with pytest.raises(NotImplementedError): protocol.check_alpn() class TestPerformServerConnectionPreface(net_tservers.ServerTestBase): class handler(tcp.BaseHandler): def handle(self): # send magic self.wfile.write(codecs.decode('505249202a20485454502f322e300d0a0d0a534d0d0a0d0a', 'hex_codec')) self.wfile.flush() # send empty settings frame self.wfile.write(codecs.decode('000000040000000000', 'hex_codec')) self.wfile.flush() # check empty settings frame raw = http2.read_raw_frame(self.rfile) assert raw == codecs.decode('00000c040000000000000200000000000300000001', 'hex_codec') # check settings acknowledgement raw = http2.read_raw_frame(self.rfile) assert raw == codecs.decode('000000040100000000', 'hex_codec') # send settings acknowledgement self.wfile.write(codecs.decode('000000040100000000', 'hex_codec')) self.wfile.flush() def test_perform_server_connection_preface(self): c = tcp.TCPClient(("127.0.0.1", self.port)) with c.connect(): protocol = HTTP2StateProtocol(c) assert not protocol.connection_preface_performed protocol.perform_server_connection_preface() assert protocol.connection_preface_performed with pytest.raises(exceptions.TcpDisconnect): protocol.perform_server_connection_preface(force=True) class TestPerformClientConnectionPreface(net_tservers.ServerTestBase): class handler(tcp.BaseHandler): def handle(self): # check magic assert self.rfile.read(24) == HTTP2StateProtocol.CLIENT_CONNECTION_PREFACE # check empty settings frame assert self.rfile.read(9) ==\ codecs.decode('000000040000000000', 'hex_codec') # send empty settings frame self.wfile.write(codecs.decode('000000040000000000', 'hex_codec')) self.wfile.flush() # check settings acknowledgement assert self.rfile.read(9) == \ codecs.decode('
From 0d37e6edc09c99e683dd91ca0e83bbc0df8477b3 Mon Sep 17 00:00:00 2001
From: Felix Fietkau <nbd@nbd.name>
Date: Sun, 16 Jul 2017 16:56:10 +0200
Subject: lib: add uevent_next_seqnum()

Signed-off-by: Felix Fietkau <nbd@nbd.name>
---
 include/linux/kobject.h |  5 +++++
 lib/kobject_uevent.c    | 37 +++++++++++++++++++++++++++++++++++++
 2 files changed, 42 insertions(+)

--- a/lib/kobject_uevent.c
+++ b/lib/kobject_uevent.c
@@ -179,6 +179,18 @@ out:
 	return r;
 }
 
+u64 uevent_next_seqnum(void)
+{
+	u64 seq;
+
+	mutex_lock(&uevent_sock_mutex);
+	seq = ++uevent_seqnum;
+	mutex_unlock(&uevent_sock_mutex);
+
+	return seq;
+}
+EXPORT_SYMBOL_GPL(uevent_next_seqnum);
+
 /**
  * kobject_synth_uevent - send synthetic uevent with arguments
  *
generated by cgit v1.2.3 (git 2.25.1) at 2026-02-19 09:58:28 +0000
def test_asterisk_form(self): c = tcp.TCPClient(("127.0.0.1", self.port)) with c.connect(): c.convert_to_ssl() protocol = HTTP2StateProtocol(c, is_server=True) protocol.connection_preface_performed = True req = protocol.read_request(NotImplemented) assert req.first_line_format == "relative" assert req.method == "OPTIONS" assert req.path == "*" class TestReadRequestAbsolute(net_tservers.ServerTestBase): class handler(tcp.BaseHandler): def handle(self): self.wfile.write( codecs.decode('00001901050000000182448d9d29aee30c0e492c2a1170426366871c92585422e085', 'hex_codec')) self.wfile.flush() ssl = True def test_absolute_form(self): c = tcp.TCPClient(("127.0.0.1", self.port)) with c.connect(): c.convert_to_ssl() protocol = HTTP2StateProtocol(c, is_server=True) protocol.connection_preface_performed = True req = protocol.read_request(NotImplemented) assert req.first_line_format == "absolute" assert req.scheme == "http" assert req.host == "address" assert req.port == 22 class TestReadResponse(net_tservers.ServerTestBase): class handler(tcp.BaseHandler): def handle(self): self.wfile.write( codecs.decode('00000801040000002a88628594e78c767f', 'hex_codec')) self.wfile.write( codecs.decode('00000600010000002a666f6f626172', 'hex_codec')) self.wfile.flush() self.rfile.safe_read(9) # just to keep the connection alive a bit longer ssl = True def test_read_response(self): c = tcp.TCPClient(("127.0.0.1", self.port)) with c.connect(): c.convert_to_ssl() protocol = HTTP2StateProtocol(c) protocol.connection_preface_performed = True resp = protocol.read_response(NotImplemented, stream_id=42) assert resp.http_version == "HTTP/2.0" assert resp.status_code == 200 assert resp.reason == '' assert resp.headers.fields == ((b':status', b'200'), (b'etag', b'foobar')) assert resp.content == b'foobar' assert resp.timestamp_end class TestReadEmptyResponse(net_tservers.ServerTestBase): class handler(tcp.BaseHandler): def handle(self): self.wfile.write( codecs.decode('00000801050000002a88628594e78c767f', 'hex_codec')) self.wfile.flush() ssl = True def test_read_empty_response(self): c = tcp.TCPClient(("127.0.0.1", self.port)) with c.connect(): c.convert_to_ssl() protocol = HTTP2StateProtocol(c) protocol.connection_preface_performed = True resp = protocol.read_response(NotImplemented, stream_id=42) assert resp.stream_id == 42 assert resp.http_version == "HTTP/2.0" assert resp.status_code == 200 assert resp.reason == '' assert resp.headers.fields == ((b':status', b'200'), (b'etag', b'foobar')) assert resp.content == b'' class TestAssembleRequest: c = tcp.TCPClient(("127.0.0.1", 0)) def test_request_simple(self): bytes = HTTP2StateProtocol(self.c).assemble_request(http.Request( b'', b'GET', b'https', b'', b'', b'/', b"HTTP/2.0", (), None, )) assert len(bytes) == 1 assert bytes[0] == codecs.decode('00000d0105000000018284874188089d5c0b8170dc07', 'hex_codec') def test_request_with_stream_id(self): req = http.Request( b'', b'GET', b'https', b'', b'', b'/', b"HTTP/2.0", (), None, ) req.stream_id = 0x42 bytes = HTTP2StateProtocol(self.c).assemble_request(req) assert len(bytes) == 1 assert bytes[0] == codecs.decode('00000d0105000000428284874188089d5c0b8170dc07', 'hex_codec') def test_request_with_body(self): bytes = HTTP2StateProtocol(self.c).assemble_request(http.Request( b'', b'GET', b'https', b'', b'', b'/', b"HTTP/2.0", http.Headers([(b'foo', b'bar')]), b'foobar', )) assert len(bytes) == 2 assert bytes[0] ==\ codecs.decode('0000150104000000018284874188089d5c0b8170dc07408294e7838c767f', 'hex_codec') assert bytes[1] ==\ codecs.decode('000006000100000001666f6f626172', 'hex_codec') class TestAssembleResponse: c = tcp.TCPClient(("127.0.0.1", 0)) def test_simple(self): bytes = HTTP2StateProtocol(self.c, is_server=True).assemble_response(http.Response( b"HTTP/2.0", 200, )) assert len(bytes) == 1 assert bytes[0] ==\ codecs.decode('00000101050000000288', 'hex_codec') def test_with_stream_id(self): resp = http.Response( b"HTTP/2.0", 200, ) resp.stream_id = 0x42 bytes = HTTP2StateProtocol(self.c, is_server=True).assemble_response(resp) assert len(bytes) == 1 assert bytes[0] ==\ codecs.decode('00000101050000004288', 'hex_codec') def test_with_body(self): bytes = HTTP2StateProtocol(self.c, is_server=True).assemble_response(http.Response( b"HTTP/2.0", 200, b'', http.Headers(foo=b"bar"), b'foobar' )) assert len(bytes) == 2 assert bytes[0] ==\ codecs.decode('00000901040000000288408294e7838c767f', 'hex_codec') assert bytes[1] ==\ codecs.decode('000006000100000002666f6f626172', 'hex_codec')