Home | History | Annotate | Download | only in kernel_CryptoAPI
      1 #!/usr/bin/python
      2 #
      3 # Copyright (c) 2015 The Chromium Authors. All rights reserved.
      4 # Use of this source code is governed by a BSD-style license that can be
      5 # found in the LICENSE file.
      6 
      7 import os
      8 from autotest_lib.client.common_lib import error
      9 from autotest_lib.client.bin import test
     10 from autotest_lib.client.bin import utils
     11 from autotest_lib.client.cros import kernel_config
     12 
     13 import ctypes
     14 import hashlib
     15 import logging
     16 import binascii
     17 
     18 
     19 AF_ALG = 38
     20 SOCK_SEQPACKET = 5
     21 
     22 class sockaddr_alg(ctypes.Structure):
     23     """
     24     A python definition of the same struct from <linux/if_alg.h>
     25 
     26     struct sockaddr_alg {
     27             __u16   salg_family;
     28             __u8    salg_type[14];
     29             __u32   salg_feat;
     30             __u32   salg_mask;
     31             __u8    salg_name[64];
     32     };
     33     """
     34     _fields_ = [
     35         ('salg_family', ctypes.c_uint16),
     36         ('salg_type', ctypes.c_char * 14),
     37         ('salg_feat', ctypes.c_uint32),
     38         ('salg_mask', ctypes.c_uint32),
     39         ('salg_name', ctypes.c_char * 64),
     40     ]
     41 
     42 
     43     def __init__(self, alg_family, alg_type, alg_name, alg_feat=0, alg_mask=0):
     44         super(sockaddr_alg, self).__init__(alg_family, alg_type, alg_feat,
     45                                            alg_mask, alg_name)
     46 
     47 
     48 class kernel_CryptoAPI(test.test):
     49     """
     50     Verify that the crypto user API can't be used to load arbitrary modules.
     51     Uses the kernel module 'test_module'
     52     """
     53     version = 1
     54     preserve_srcdir = True
     55 
     56     def initialize(self):
     57         self.job.require_gcc()
     58 
     59 
     60     def setup(self):
     61         os.chdir(self.srcdir)
     62         utils.make()
     63 
     64 
     65     def try_load_mod(self, module):
     66         """
     67         Try to load a (non-crypto) module using the crypto UAPI
     68         @param module: name of the kernel module to try to load
     69         """
     70         if utils.module_is_loaded(module):
     71             utils.unload_module(module)
     72 
     73         path = os.path.join(self.srcdir, 'crypto_load_mod ')
     74         utils.system(path + module)
     75 
     76         if utils.module_is_loaded(module):
     77             utils.unload_module(module)
     78             raise error.TestFail('Able to load module "%s" using crypto UAPI' %
     79                                  module)
     80 
     81 
     82     def do_ifalg_digest(self, name, data, outlen):
     83         """
     84         Use ctypes to run a digest through one of the available kernel ifalg
     85         digest types
     86 
     87         @param name: digest name
     88         @param data: string data to digest
     89         @param outlen: length of the digest output (e.g., SHA1 is 160-bit, so
     90                 outlen==20)
     91         @param return string containing the output digest, or None if
     92                 experiencing an error
     93         """
     94         libc = ctypes.CDLL("libc.so.6", use_errno=True)
     95 
     96         # If you don't specify the function parameters this way, ctypes may try
     97         # to treat pointers as 32-bit (and then later sign-extend them)
     98         libc.socket.argtypes = [ ctypes.c_int, ctypes.c_int, ctypes.c_int ]
     99         libc.bind.argtypes = [ ctypes.c_int, ctypes.c_void_p, ctypes.c_int ]
    100         libc.send.argtypes = [ ctypes.c_int, ctypes.c_void_p, ctypes.c_size_t,
    101                                ctypes.c_int ]
    102         libc.recv.argtypes = [ ctypes.c_int, ctypes.c_void_p, ctypes.c_size_t,
    103                                ctypes.c_int ]
    104 
    105         sock = libc.socket(AF_ALG, SOCK_SEQPACKET, 0)
    106         if sock == -1:
    107             libc.perror("socket")
    108             return None
    109 
    110         alg = sockaddr_alg(AF_ALG, "hash", name)
    111         if libc.bind(sock, ctypes.addressof(alg), ctypes.sizeof(alg)) == -1:
    112             libc.perror("bind")
    113             return None
    114 
    115         fd = libc.accept(sock, None, 0)
    116         if fd == -1:
    117             libc.perror("accept")
    118             return None
    119 
    120         message = ctypes.create_string_buffer(data, len(data))
    121         if libc.send(fd, ctypes.addressof(message), ctypes.sizeof(message), 0) == -1:
    122             libc.perror("send")
    123             return None
    124 
    125         out = (ctypes.c_uint8 * outlen)()
    126         ret = libc.recv(fd, ctypes.addressof(out), ctypes.sizeof(out), 0)
    127         if ret == -1:
    128             libc.perror("recv")
    129             return None
    130 
    131         h = ctypes.string_at(ctypes.addressof(out), ret)
    132 
    133         libc.close(sock)
    134 
    135         return h
    136 
    137 
    138     def test_digest(self, name, lib, data):
    139         """
    140         Run a digest through both the kernel UAPI and through hashlib, throwing
    141         an error if the two don't match
    142 
    143         @param name: name of the digest (according to AF_ALG)
    144         @param lib: a hashlib digest object
    145         @param data: data to digest
    146         """
    147 
    148         logging.info("Testing digest %s", name)
    149 
    150         h1 = self.do_ifalg_digest(name, data, lib.digestsize)
    151         if h1 is None:
    152             raise error.TestFail("ifalg digest %s failed", name)
    153 
    154         lib.update(data)
    155         h2 = lib.digest()
    156 
    157         if h1 != h2:
    158             logging.error("%s: digests do not match", name)
    159             logging.error(" hash 1: %s", binascii.hexlify(h1))
    160             logging.error(" hash 2: %s", binascii.hexlify(h2))
    161             raise error.TestFail("digest mismatch (%s)" % name)
    162 
    163         logging.debug("hash 1: %s", binascii.hexlify(h1))
    164         logging.debug("hash 2: %s", binascii.hexlify(h2))
    165 
    166 
    167     def test_digests(self, data):
    168         """
    169         Test several digests, using both the kernel crypto APIs and python
    170         hashlib
    171 
    172         @param data: the data to digest
    173         """
    174 
    175         digests = [
    176                 ( "sha1", hashlib.sha1()),
    177                 ( "md5", hashlib.md5()),
    178                 ( "sha512", hashlib.sha512()),
    179         ]
    180 
    181         for (name, lib) in digests:
    182             self.test_digest(name, lib, data)
    183 
    184 
    185     def test_is_valid(self):
    186         """
    187         Check if this test is worth running, based on whether the kernel
    188         .config has the right features
    189         """
    190         config = kernel_config.KernelConfig()
    191         config.initialize()
    192         config.is_enabled('CRYPTO_USER_API_HASH')
    193         config.is_enabled('CRYPTO_USER_API')
    194         return len(config.failures()) == 0
    195 
    196 
    197     def run_once(self):
    198         # crypto tests only work with AF_ALG support
    199         if not self.test_is_valid():
    200             raise error.TestNAError("Crypto tests only run with AF_ALG support")
    201 
    202         module = "test_module"
    203         self.try_load_mod(module)
    204 
    205         self.test_digests("This is a not-so-secret message")
    206