64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
|
"""Bounds/replay checks for untrusted cursor metadata before SDL allocation."""
|
||
|
|
import base64
|
||
|
|
import importlib.util
|
||
|
|
import hashlib
|
||
|
|
from pathlib import Path
|
||
|
|
import unittest
|
||
|
|
import tempfile
|
||
|
|
|
||
|
|
spec = importlib.util.spec_from_file_location(
|
||
|
|
'native_cursor_client', Path(__file__).resolve().parents[2] /
|
||
|
|
'terminal/native-cursor/client.py')
|
||
|
|
client = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(client)
|
||
|
|
|
||
|
|
|
||
|
|
class CursorProtocolTests(unittest.TestCase):
|
||
|
|
def state(self):
|
||
|
|
return dict(type='cursor', version=1, lease='c'*32, seq=2, ack=0, w=2, h=2,
|
||
|
|
hx=1, hy=1, x=100, y=100, sw=2560, sh=1440,
|
||
|
|
rgba=base64.b64encode(bytes([255]*16)).decode())
|
||
|
|
|
||
|
|
def test_valid_and_hidden(self):
|
||
|
|
for pixels in (bytes([255]*16), bytes(16)):
|
||
|
|
s = self.state()
|
||
|
|
s['rgba'] = base64.b64encode(pixels).decode()
|
||
|
|
packet = client.validate(s, 'c'*32, [2560, 1440], 1)
|
||
|
|
self.assertEqual(len(packet), 88)
|
||
|
|
self.assertEqual(packet[64:80], pixels)
|
||
|
|
|
||
|
|
def test_bad_metadata(self):
|
||
|
|
for key, value in [('version', 2), ('version', True), ('ack', -1),
|
||
|
|
('ack', 2**63), ('lease', 'd'*32), ('type', 'ready'),
|
||
|
|
('seq', 1), ('seq', 2**63), ('w', 0), ('w', 129),
|
||
|
|
('h', -1), ('hx', 2), ('hy', -1), ('x', -1),
|
||
|
|
('y', 1440), ('sw', 1920), ('w', True),
|
||
|
|
('seq', 2.0), ('rgba', '***'), ('rgba', 'AAAA')]:
|
||
|
|
with self.subTest(key=key, value=value):
|
||
|
|
state = self.state()
|
||
|
|
state[key] = value
|
||
|
|
with self.assertRaises(ValueError):
|
||
|
|
client.validate(state, 'c'*32, [2560, 1440], 1)
|
||
|
|
|
||
|
|
def test_missing_field(self):
|
||
|
|
state = self.state()
|
||
|
|
del state['h']
|
||
|
|
with self.assertRaises(ValueError):
|
||
|
|
client.validate(state, 'c'*32, [2560, 1440], 1)
|
||
|
|
|
||
|
|
def test_old_client_rejected_before_lease(self):
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
binary = Path(directory) / 'client'
|
||
|
|
binary.write_bytes(b'validated client')
|
||
|
|
expected = hashlib.sha256(binary.read_bytes()).hexdigest()
|
||
|
|
client.verify_client(binary, expected)
|
||
|
|
binary.write_bytes(b'old client without cursor receiver')
|
||
|
|
with self.assertRaises(ValueError):
|
||
|
|
client.verify_client(binary, expected)
|
||
|
|
with self.assertRaises(ValueError):
|
||
|
|
client.verify_client(binary, None)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
unittest.main()
|