File: umac.py

package info (click to toggle)
python-asyncssh 2.21.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 3,464 kB
  • sloc: python: 40,306; makefile: 11
file content (139 lines) | stat: -rw-r--r-- 4,391 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
# Copyright (c) 2016-2024 by Ron Frederick <ronf@timeheart.net> and others.
#
# This program and the accompanying materials are made available under
# the terms of the Eclipse Public License v2.0 which accompanies this
# distribution and is available at:
#
#     http://www.eclipse.org/legal/epl-2.0/
#
# This program may also be made available under the following secondary
# licenses when the conditions for such availability set forth in the
# Eclipse Public License v2.0 are satisfied:
#
#    GNU General Public License, Version 2.0, or any later versions of
#    that license
#
# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
#
# Contributors:
#     Ron Frederick - initial implementation, API, and documentation

"""UMAC cryptographic hash (RFC 4418) wrapper for Nettle library"""

import binascii
import ctypes
import ctypes.util
from typing import TYPE_CHECKING, Callable, Optional


if TYPE_CHECKING:
    _ByteArray = ctypes.Array[ctypes.c_char]
    _SetKey = Callable[[_ByteArray, bytes], None]
    _SetNonce = Callable[[_ByteArray, ctypes.c_size_t, bytes], None]
    _Update = Callable[[_ByteArray, ctypes.c_size_t, bytes], None]
    _Digest = Callable[[_ByteArray, ctypes.c_size_t, _ByteArray], None]
    _New = Callable[[bytes, Optional[bytes], Optional[bytes]], object]


_UMAC_BLOCK_SIZE = 1024
_UMAC_DEFAULT_CTX_SIZE = 4096


def _build_umac(size: int) -> '_New':
    """Function to build UMAC wrapper for a specific digest size"""

    _name = f'umac{size}'
    _prefix = f'nettle_{_name}_'

    try:
        _context_size: int = getattr(_nettle, _prefix + '_ctx_size')()
    except AttributeError:
        _context_size = _UMAC_DEFAULT_CTX_SIZE

    _set_key: _SetKey = getattr(_nettle, _prefix + 'set_key')
    _set_nonce: _SetNonce = getattr(_nettle, _prefix + 'set_nonce')
    _update: _Update = getattr(_nettle, _prefix + 'update')
    _digest: _Digest = getattr(_nettle, _prefix + 'digest')


    class _UMAC:
        """Wrapper for UMAC cryptographic hash

           This class supports the cryptographic hash API defined in PEP 452.

        """

        name = _name
        block_size = _UMAC_BLOCK_SIZE
        digest_size = size // 8

        def __init__(self, ctx: '_ByteArray', nonce: Optional[bytes] = None,
                     msg: Optional[bytes] = None):
            self._ctx = ctx

            if nonce:
                self.set_nonce(nonce)

            if msg:
                self.update(msg)

        @classmethod
        def new(cls, key: bytes, msg: Optional[bytes] = None,
                nonce: Optional[bytes] = None) -> '_UMAC':
            """Construct a new UMAC hash object"""

            ctx = ctypes.create_string_buffer(_context_size)
            _set_key(ctx, key)

            return cls(ctx, nonce, msg)

        def copy(self) -> '_UMAC':
            """Return a new hash object with this object's state"""

            ctx = ctypes.create_string_buffer(self._ctx.raw)
            return self.__class__(ctx)

        def set_nonce(self, nonce: bytes) -> None:
            """Reset the nonce associated with this object"""

            _set_nonce(self._ctx, ctypes.c_size_t(len(nonce)), nonce)

        def update(self, msg: bytes) -> None:
            """Add the data in msg to the hash"""

            _update(self._ctx, ctypes.c_size_t(len(msg)), msg)

        def digest(self) -> bytes:
            """Return the hash and increment nonce to begin a new message

               .. note:: The hash is reset and the nonce is incremented
                         when this function is called. This doesn't match
                         the behavior defined in PEP 452.

            """

            result = ctypes.create_string_buffer(self.digest_size)
            _digest(self._ctx, ctypes.c_size_t(self.digest_size), result)
            return result.raw

        def hexdigest(self) -> str:
            """Return the digest as a string of hexadecimal digits"""

            return binascii.b2a_hex(self.digest()).decode('ascii')


    return _UMAC.new


for lib in ('nettle', 'libnettle', 'libnettle-6'):
    _nettle_lib = ctypes.util.find_library(lib)

    if _nettle_lib: # pragma: no branch
        break
else: # pragma: no cover
    _nettle_lib = None

if _nettle_lib: # pragma: no branch
    _nettle = ctypes.cdll.LoadLibrary(_nettle_lib)

    umac32, umac64, umac96, umac128 = map(_build_umac, (32, 64, 96, 128))