tests/test_setdatafd.py

   1 #!/usr/bin/python
   2
   3 import os
   4 import sys
   5 import socket
   6 import subprocess
   7 import time
   8 import struct
   9
  10 from array import array
  11 if sys.version_info[0] < 3:
  12     import twisted.python.sendmsg as sendmsg
  13
  14
  15 def toString(arr):
  16     return ' '.join('{:02x}'.format(x) for x in arr)
  17
  18
  19 def test_ReadPCR10(fd):
  20     send_data = bytearray(b"\x00\xC1\x00\x00\x00\x0C\x00\x00\x00\x99\x00\x01")
  21     exp_data = bytearray([0x00, 0xC4, 0x00, 0x00, 0x00, 0x0A,
  22                           0x00, 0x00, 0x00, 0x26])
  23
  24     try:
  25         print("Sending data over ....")
  26         n = fd.send(send_data)
  27         print("Written %d bytes " % n)
  28     except socket.error as e:
  29         print("SocketError")
  30         fd.close()
  31         return False
  32
  33     buf = fd.recv(1024)
  34     fd.close()
  35     if buf:
  36         if bytearray(buf) == exp_data:
  37             return True
  38         else:
  39             print("Unexpected reply:\n  actual: %s\n  expected: %s"
  40                   % (toString(buf), toString(exp_data)))
  41             return False
  42     else:
  43         print("Null reply from swtpm")
  44         return False
  45
  46
  47 def test_SetDatafd():
  48     fd, _fd = socket.socketpair(socket.AF_UNIX, socket.SOCK_DGRAM)
  49     sock_path = os.getenv('SOCK_PATH')
  50     cmd_set_data_fd = bytearray([0x00, 0x00, 0x00, 0x10])
  51     expected_res = bytearray([0x00, 0x00, 0x00, 0x00])
  52     try:
  53         fds = array("i")
  54         fds.append(_fd.fileno())
  55         ctrlfd = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  56         print("Connecting to server at : %s" % sock_path)
  57         ctrlfd.connect(sock_path)
  58         print("Sending data fd over ctrl fd...")
  59         if sys.version_info[0] < 3:
  60             sendmsg.send1msg(ctrlfd.fileno(), str(cmd_set_data_fd), 0,
  61                              [(socket.SOL_SOCKET,
  62                                sendmsg.SCM_RIGHTS,
  63                                struct.pack("i", _fd.fileno()))])
  64         else:
  65             ctrlfd.sendmsg([cmd_set_data_fd],
  66                            [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fds)])
  67     except socket.error as e:
  68         print("SocketError: " + str(e))
  69         ctrlfd.close()
  70
  71     buf = ctrlfd.recv(4)
  72     print("Received bytes.. : %s" % buf)
  73     if buf:
  74         caps = bytearray(buf)
  75         if caps == expected_res:
  76             return test_ReadPCR10(fd)
  77         else:
  78             print("Unexpected reply for CMD_SET_DATA_FD: \n"
  79                   "  actual: %s\n  expected: %s"
  80                   % (toString(caps), toString(expected_res)))
  81             return False
  82     else:
  83         print("Null reply from swtpm")
  84         return False
  85
  86 if __name__ == "__main__":
  87     try:
  88         if not test_SetDatafd():
  89             res = 1
  90         else:
  91             res = 0
  92     except:
  93         print("__Exception: ", sys.exc_info())
  94         res = -1
  95
  96     sys.exit(res)