# dictionary.py
#
# Copyright 2002,2005,2007,2016 Wichert Akkerman <wichert@wiggy.net>
"""
RADIUS uses dictionaries to define the attributes that can
be used in packets. The Dictionary class stores the attribute
definitions from one or more dictionary files in the FreeRADIUS format.
Dictionary files are text files with one command per line.
Comments are specified by starting with a # character, and empty
lines are ignored.
The commands supported are:
``ATTRIBUTE <attribute> <code> <type> [<vendor>|<flags>]``
Define an attribute and its data type. The code is decimal, or
hexadecimal with a ``0x`` prefix (octal with ``0o``), and
``<tlv code>.<code>`` for a sub-attribute of a TLV.
``VALUE <attribute> <valuename> <value>``
Define a named value of an attribute.
``VENDOR <name> <id> [format=<t>,<l>]``
Define a vendor and its ID. Only the RFC 2865 format ``format=1,1`` (the
default) is supported, the attributes of vendors with another format
are skipped.
``BEGIN-VENDOR <name> [format=<attribute>]``
Begin the definition of the attributes of a vendor. The attributes of a
block with a ``format=`` option are extended vendor-specific attributes
(RFC 6929) and are skipped.
``END-VENDOR <name>``
End the definition of the attributes of a vendor.
``$INCLUDE <filename>``
Read another dictionary file. A relative file name is relative to the
directory of the current file, or to the current working directory if
the dictionary is read from a file-like object.
``$INCLUDE- <filename>``
Like ``$INCLUDE``, but a file that doesn't exist is ignored.
Other commands are ignored.
The attribute flags are a comma separated list of:
``has_tag``
The attribute is tagged (RFC 2868).
``encrypt=1``
User-Password encryption (RFC 2865 section 5.2). It is not applied
automatically, see :meth:`pyrad.packet.AuthPacket.PwCrypt` and
:meth:`pyrad.packet.AuthPacket.PwDecrypt`.
``encrypt=2``
Salt encryption (RFC 2868 section 3.5), applied automatically.
``encrypt=3``
Ascend-Send-Secret encryption. Accepted, but not applied.
The FreeRADIUS flags ``secret``, ``virtual`` and ``concat`` are accepted
and ignored.
FreeRADIUS dictionaries also define attributes that pyrad cannot encode or
decode. These are skipped with a warning, so that the FreeRADIUS dictionary
files can be read as they are:
- attributes of the data types ``vsa``, ``extended``, ``long-extended``,
``evs`` (the RFC 6929 extended attributes), ``ipv4prefix``,
``combo-ip``, ``bool``, ``uint16`` and ``uint32``, and their
sub-attributes
- attributes with the ``array`` flag
- TLVs nested more than one level deep
- attributes of vendors with a format other than ``format=1,1``
(``VENDOR``)
- vendor attributes in a ``BEGIN-VENDOR`` block with a ``format=``
option, which are extended vendor-specific attributes (RFC 6929)
VALUE definitions for skipped attributes are ignored, and VALUE definitions
for unknown attributes are ignored with a warning. Received attributes
that were skipped are decoded as undefined attributes.
The data types currently supported, and the Python values used for them in
packets:
.. list-table::
:header-rows: 1
:widths: 15 40 45
* - type
- description
- Python value
* - string
- UTF-8 string
- :class:`str` or :class:`bytes`, at most 253 octets encoded; a
received value which is not valid UTF-8 is returned as
:class:`bytes`
* - ipaddr
- IPv4 address
- :class:`str`, e.g. ``'192.0.2.1'``
* - date
- 32 bits UNIX timestamp
- :class:`int`
* - octets
- arbitrary binary data
- :class:`bytes`; a value starting with ``0x`` is decoded from hex
* - abinary
- Ascend binary filter
- Ascend filter :class:`str`, e.g. ``'family=ipv4 action=accept
direction=in'``; received values are returned as :class:`bytes`
* - ipv6addr
- 16 octets in network byte order
- :class:`str` or :class:`ipaddress.IPv6Address`; received values
are returned as :class:`str`
* - ipv6prefix
- up to 18 octets in network byte order
- :class:`str` ``'address/length'``, e.g. ``'2001:db8::/32'``
* - integer
- 32 bits unsigned number
- :class:`int`
* - signed
- 32 bits signed number
- :class:`int`
* - short
- 16 bits unsigned number
- :class:`int`
* - byte
- 8 bits unsigned number
- :class:`int`
* - integer64
- 64 bits unsigned number
- :class:`int`
* - tlv
- nested type-length-value
- set the sub-attributes by their names; reading the TLV returns a
:class:`dict` mapping sub-attribute names to lists of values
* - ifid
- 8 octets IPv6 interface identifier
- :class:`str` ``'hhhh:hhhh:hhhh:hhhh'``, ``h`` being a hex digit
* - ether
- 6 octets Ethernet MAC address
- :class:`str` ``'hh:hh:hh:hh:hh:hh'``, ``h`` being a hex digit,
upper or lowercase
"""
import logging
import struct
from pyrad import bidict
from pyrad import tools
from pyrad import dictfile
from copy import copy
__docformat__ = "restructuredtext en"
logger = logging.getLogger('pyrad')
DATATYPES = frozenset(['string', 'ipaddr', 'integer', 'date', 'octets',
'abinary', 'ipv6addr', 'ipv6prefix', 'short', 'byte',
'signed', 'ifid', 'ether', 'tlv', 'integer64'])
# FreeRADIUS data types which pyrad does not support. Attributes of these
# types are skipped with a warning instead of failing the parse.
UNSUPPORTED_DATATYPES = frozenset(['vsa', 'extended', 'long-extended', 'evs',
'ipv4prefix', 'combo-ip', 'bool', 'uint16',
'uint32'])
# Attribute flags; a fifth ATTRIBUTE column that contains one of them is a
# list of flags instead of a vendor name.
ATTRIBUTE_FLAGS = frozenset(['has_tag', 'encrypt', 'secret', 'array',
'virtual', 'concat'])
[docs]
class ParseError(Exception):
"""A dictionary file can't be parsed.
Raised for syntax errors and recursive includes. Note that it is not a
subclass of :exc:`ValueError`.
:ivar msg: error message
:vartype msg: str
:ivar file: name of the dictionary file, or ``''``
:vartype file: str
:ivar line: line number on which the error occurred, or -1
:vartype line: int
"""
def __init__(self, msg=None, **data):
self.msg = msg
self.file = data.get('file', '')
self.line = data.get('line', -1)
def __str__(self):
str = ''
if self.file:
str += self.file
if self.line > -1:
str += '(%d)' % self.line
if self.file or self.line > -1:
str += ': '
str += 'Parse error'
if self.msg:
str += ': %s' % self.msg
return str
[docs]
class Attribute(object):
"""The definition of an attribute in a :class:`Dictionary`.
Raises :exc:`ValueError` for an unsupported data type.
:ivar name: name of the attribute
:vartype name: str
:ivar code: attribute code, for a sub-attribute its code in the TLV
:vartype code: int
:ivar type: data type, e.g. ``'string'``
:vartype type: str
:ivar vendor: name of the vendor, ``''`` for a standard attribute
:vartype vendor: str
:ivar encrypt: encryption method of the ``encrypt=`` flag, 0 for none
:vartype encrypt: int
:ivar has_tag: the attribute is tagged (RFC 2868)
:vartype has_tag: bool
:ivar values: named values, mapping the name to the encoded value
:vartype values: pyrad.bidict.BiDict
:ivar sub_attributes: names of the sub-attributes of a TLV by their
code
:vartype sub_attributes: dict[int, str]
:ivar parent: the TLV of a sub-attribute, else None
:vartype parent: Attribute or None
:ivar is_sub_attribute: the attribute is a sub-attribute of a TLV
:vartype is_sub_attribute: bool
"""
def __init__(self, name, code, datatype, is_sub_attribute=False, vendor='', values=None,
encrypt=0, has_tag=False):
if datatype not in DATATYPES:
raise ValueError('Invalid data type')
self.name = name
self.code = code
self.type = datatype
self.vendor = vendor
self.encrypt = encrypt
self.has_tag = has_tag
self.values = bidict.BiDict()
self.sub_attributes = {}
self.parent = None
self.is_sub_attribute = is_sub_attribute
if values:
for (key, value) in values.items():
self.values.Add(key, value)
[docs]
class Dictionary:
"""RADIUS dictionary.
Stores the vendors, attributes and their values defined in RADIUS
dictionary files. ``Dictionary(path, *more)`` reads the files in
order, see :meth:`ReadDictionary`. ``dictionary['User-Name']`` returns
the :class:`Attribute` of that name and ``'User-Name' in dictionary``
checks whether it is defined.
:ivar attributes: attributes by name
:vartype attributes: dict[str, Attribute]
:ivar attrindex: attribute names and the keys of the attributes in a
packet: the code, ``(vendor, code)`` for a vendor attribute,
``(tlv code, code)`` for a sub-attribute and
``(vendor, tlv code, code)`` for a sub-attribute of a vendor TLV,
with the vendor ID as vendor
:vartype attrindex: pyrad.bidict.BiDict
:ivar vendors: vendor names and IDs, ``''`` is vendor 0
:vartype vendors: pyrad.bidict.BiDict
:ivar skipped_attributes: names of the attributes that were skipped
because pyrad does not support them; using them raises a
:exc:`KeyError`
:vartype skipped_attributes: set[str]
"""
def __init__(self, dict=None, *dicts):
"""
:param dict: path of a dictionary file or file-like object to read
:type dict: str or io.TextIOBase
:param dicts: further dictionary files to read
:type dicts: str or io.TextIOBase
"""
self.vendors = bidict.BiDict()
self.vendors.Add('', 0)
self.attrindex = bidict.BiDict()
self.attributes = {}
self.defer_parse = []
self.skipped_attributes = set()
# vendors whose attribute format is not the RFC 2865 one
self._unsupported_vendors = set()
# TLVs by (vendor code, attribute code), kept across ReadDictionary
# calls so that sub-attributes can be defined in another file
self._tlvs = {}
# (vendor code, attribute code) of skipped top-level attributes,
# whose sub-attributes are skipped as well
self._skipped_codes = set()
if dict:
self.ReadDictionary(dict)
for i in dicts:
self.ReadDictionary(i)
def __len__(self):
return len(self.attributes)
def __getitem__(self, key):
return self.attributes[key]
def __contains__(self, key):
return key in self.attributes
has_key = __contains__
def __SkipAttribute(self, state, name, reason):
self.skipped_attributes.add(name)
state['skipped'].append(reason)
logger.debug('%s(%d): skipping attribute %s: %s',
state['file'], state['line'], name, reason)
def __ParseAttribute(self, state, tokens):
if len(tokens) not in [4, 5]:
raise ParseError(
'Incorrect number of tokens for attribute definition',
file=state['file'],
line=state['line'])
vendor = state['vendor']
has_tag = False
encrypt = 0
array = False
if len(tokens) >= 5:
def keyval(o):
kv = o.split('=')
if len(kv) == 2:
return (kv[0], kv[1])
else:
return (kv[0], None)
options = [keyval(o) for o in tokens[4].split(',')]
is_flags = any(key in ATTRIBUTE_FLAGS for (key, val) in options)
if self.vendors.HasForward(tokens[4]) and \
not any(key in ('has_tag', 'encrypt')
for (key, val) in options):
vendor = tokens[4]
elif not is_flags:
raise ParseError('Unknown vendor ' + tokens[4],
file=state['file'],
line=state['line'])
for (key, val) in options if is_flags else []:
if key == 'has_tag':
has_tag = True
elif key == 'encrypt':
if val not in ['1', '2', '3']:
raise ParseError(
'Illegal attribute encryption: %s' % val,
file=state['file'],
line=state['line'])
encrypt = int(val)
elif key == 'array':
array = True
(attribute, code, datatype) = tokens[1:4]
codes = code.split('.')
# Codes can be sent as hex, or octal or decimal string representations.
tmp = []
try:
for c in codes:
if c.startswith('0x'):
tmp.append(int(c, 16))
elif c.startswith('0o'):
tmp.append(int(c, 8))
else:
tmp.append(int(c, 10))
except ValueError:
raise ParseError('Illegal attribute code: ' + code,
file=state['file'],
line=state['line'])
codes = tmp
datatype = datatype.split("[")[0]
if datatype not in DATATYPES and datatype not in UNSUPPORTED_DATATYPES:
raise ParseError('Illegal type: ' + datatype,
file=state['file'],
line=state['line'])
vendor_code = self.vendors.GetForward(vendor)
if len(codes) > 1 and (vendor_code, codes[0]) in self._skipped_codes:
return self.__SkipAttribute(
state, attribute, 'sub-attribute of unsupported attribute')
if state['vendor_format']:
return self.__SkipAttribute(
state, attribute, 'extended vendor-specific attribute')
if vendor in self._unsupported_vendors:
return self.__SkipAttribute(state, attribute, 'vendor format')
if datatype in UNSUPPORTED_DATATYPES:
if len(codes) == 1:
self._skipped_codes.add((vendor_code, codes[0]))
self._tlvs.pop((vendor_code, codes[0]), None)
return self.__SkipAttribute(state, attribute,
'data type ' + datatype)
if array:
return self.__SkipAttribute(state, attribute, 'array flag')
if len(codes) > 2 or (len(codes) == 2 and datatype == 'tlv'):
return self.__SkipAttribute(state, attribute, 'nested TLV')
is_sub_attribute = (len(codes) > 1)
if len(codes) == 2:
code = int(codes[1])
parent_code = int(codes[0])
parent = self._tlvs.get((vendor_code, parent_code))
if parent is None:
raise ParseError(
'Sub-attribute %s of unknown TLV %d' % (attribute,
parent_code),
file=state['file'],
line=state['line'])
else:
code = int(codes[0])
parent_code = None
if vendor:
if is_sub_attribute:
key = (vendor_code, parent_code, code)
else:
key = (vendor_code, code)
else:
if is_sub_attribute:
key = (parent_code, code)
else:
key = code
self.attrindex.Add(attribute, key)
self.attributes[attribute] = Attribute(attribute, code, datatype, is_sub_attribute, vendor, encrypt=encrypt, has_tag=has_tag)
if is_sub_attribute:
# save sub attribute in parent tlv and update their parent field
parent.sub_attributes[code] = attribute
self.attributes[attribute].parent = parent
else:
# a (re)defined top-level attribute replaces a skipped attribute
# or TLV with the same code
self._skipped_codes.discard((vendor_code, code))
if datatype == 'tlv':
self._tlvs[(vendor_code, code)] = self.attributes[attribute]
else:
self._tlvs.pop((vendor_code, code), None)
def __ParseValue(self, state, tokens, defer):
if len(tokens) != 4:
raise ParseError('Incorrect number of tokens for value definition',
file=state['file'],
line=state['line'])
(attr, key, value) = tokens[1:]
if attr in self.skipped_attributes and attr not in self.attributes:
return
try:
adef = self.attributes[attr]
except KeyError:
if defer:
self.defer_parse.append((copy(state), copy(tokens)))
return
# FreeRADIUS dictionaries contain a few of these, so warn
# instead of failing the parse
logger.warning('%s(%d): ignoring value defined for unknown '
'attribute %s', state['file'], state['line'], attr)
return
try:
if adef.type in ['integer', 'signed', 'short', 'byte',
'integer64', 'date']:
value = int(value, 0)
value = tools.EncodeAttr(adef.type, value)
except (ValueError, TypeError, struct.error) as e:
raise ParseError('Illegal value %s for attribute %s: %s'
% (tokens[3], attr, e),
file=state['file'],
line=state['line'])
self.attributes[attr].values.Add(key, value)
def __ParseVendor(self, state, tokens):
if len(tokens) not in [3, 4]:
raise ParseError(
'Incorrect number of tokens for vendor definition',
file=state['file'],
line=state['line'])
(vendorname, vendor) = tokens[1:3]
# pyrad only encodes the RFC 2865 vendor attribute format (1 octet
# type, 1 octet length), attributes of vendors with another format
# are skipped
if len(tokens) == 4:
fmt = tokens[3].split('=')
if fmt[0] != 'format':
raise ParseError(
"Unknown option '%s' for vendor definition" % (fmt[0]),
file=state['file'],
line=state['line'])
try:
fmtargs = fmt[1].split(',')
# ",c": WiMAX continuation octet
continuation = len(fmtargs) == 3 and fmtargs[2] == 'c'
if continuation:
fmtargs = fmtargs[:2]
(token, length) = tuple(int(a) for a in fmtargs)
except ValueError:
raise ParseError(
'Syntax error in vendor specification',
file=state['file'],
line=state['line'])
if token not in [1, 2, 4] or length not in [0, 1, 2]:
raise ParseError(
'Unknown vendor format specification %s' % (fmt[1]),
file=state['file'],
line=state['line'])
if (token, length, continuation) != (1, 1, False):
self._unsupported_vendors.add(vendorname)
else:
self._unsupported_vendors.discard(vendorname)
try:
vendor_code = int(vendor, 0)
except ValueError:
raise ParseError('Illegal vendor code: ' + vendor,
file=state['file'],
line=state['line'])
self.vendors.Add(vendorname, vendor_code)
def __ParseBeginVendor(self, state, tokens):
if len(tokens) not in [2, 3]:
raise ParseError(
'Incorrect number of tokens for begin-vendor statement',
file=state['file'],
line=state['line'])
# BEGIN-VENDOR <vendor> format=<Extended-Vendor-Specific attribute>
# defines extended vendor-specific attributes (RFC 6929), which are
# skipped
vendor_format = None
if len(tokens) == 3:
if not tokens[2].startswith('format='):
raise ParseError(
"Unknown option '%s' for begin-vendor statement" %
tokens[2],
file=state['file'],
line=state['line'])
vendor_format = tokens[2][len('format='):]
vendor = tokens[1]
if not self.vendors.HasForward(vendor):
raise ParseError(
'Unknown vendor %s in begin-vendor statement' % vendor,
file=state['file'],
line=state['line'])
state['vendor'] = vendor
state['vendor_format'] = vendor_format
def __ParseEndVendor(self, state, tokens):
if len(tokens) != 2:
raise ParseError(
'Incorrect number of tokens for end-vendor statement',
file=state['file'],
line=state['line'])
vendor = tokens[1]
if state['vendor'] != vendor:
raise ParseError(
'Ending non-open vendor ' + vendor,
file=state['file'],
line=state['line'])
state['vendor'] = ''
state['vendor_format'] = None
[docs]
def ReadDictionary(self, file):
"""Read a dictionary file.
Its definitions are merged into the dictionary; later definitions
replace earlier ones. Unsupported attributes are skipped, listed in
``skipped_attributes``, and reported with one warning on the
``pyrad`` logger.
:param file: path of a dictionary file or a file-like object
:type file: str or io.TextIOBase
:raises ParseError: for a syntax error or a recursive include
:raises OSError: if a file can't be opened
"""
fil = dictfile.DictFile(file)
state = {}
state['vendor'] = ''
state['vendor_format'] = None
state['skipped'] = []
self.defer_parse = []
for line in fil:
state['file'] = fil.File()
state['line'] = fil.Line()
line = line.split('#', 1)[0].strip()
tokens = line.split()
if not tokens:
continue
key = tokens[0].upper()
if key == 'ATTRIBUTE':
self.__ParseAttribute(state, tokens)
elif key == 'VALUE':
self.__ParseValue(state, tokens, True)
elif key == 'VENDOR':
self.__ParseVendor(state, tokens)
elif key == 'BEGIN-VENDOR':
self.__ParseBeginVendor(state, tokens)
elif key == 'END-VENDOR':
self.__ParseEndVendor(state, tokens)
for state, tokens in self.defer_parse:
key = tokens[0].upper()
if key == 'VALUE':
self.__ParseValue(state, tokens, False)
self.defer_parse = []
if state['skipped']:
reasons = sorted(set(state['skipped']))
logger.warning('Skipped %d unsupported dictionary attributes (%s)',
len(state['skipped']), ', '.join(reasons))