rasdaman-upstream.git - Cleaning up upstream's build for packaging (git://kahlua.eecs.jacobs-university.de/rasdaman.git)

diff options

Diffstat (limited to 'src/grammar/ComplexConst.java')

-rw-r--r--

1 files changed, 0 insertions, 59 deletions

# Authors: # Jason Gerard DeRose <jderose@redhat.com> # Rob Crittenden <rcritten@redhat.com> # # Copyright (C) 2008 Red Hat # see file 'COPYING' for use and warranty information # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. """ RPC client and shared RPC client/server functionality. This module adds some additional functionality on top of the ``xmlrpc.client`` module in the Python standard library (``xmlrpclib`` in Python 2). For documentation on the ``xmlrpclib`` module, see: http://docs.python.org/2/library/xmlrpclib.html Also see the `ipaserver.rpcserver` module. """ from __future__ import absolute_import from decimal import Decimal import datetime import os import locale import base64 import json import socket import gzip import gssapi from dns import resolver, rdatatype from dns.exception import DNSException from ssl import SSLError import six from six.moves import urllib from ipalib.backend import Connectible from ipalib.constants import LDAP_GENERALIZED_TIME_FORMAT from ipalib.errors import (public_errors, UnknownError, NetworkError, KerberosError, XMLRPCMarshallError, JSONError) from ipalib import errors, capabilities from ipalib.request import context, Connection from ipapython.ipa_log_manager import root_logger from ipapython import ipautil from ipapython import session_storage from ipapython.cookie import Cookie from ipapython.dnsutil import DNSName from ipalib.text import _ from ipalib.util import create_https_connection from ipalib.krb_utils import KRB5KDC_ERR_S_PRINCIPAL_UNKNOWN, KRB5KRB_AP_ERR_TKT_EXPIRED, \ KRB5_FCC_PERM, KRB5_FCC_NOFILE, KRB5_CC_FORMAT, \ KRB5_REALM_CANT_RESOLVE, KRB5_CC_NOTFOUND, get_principal from ipapython.dn import DN from ipapython.kerberos import Principal from ipalib.capabilities import VERSION_WITHOUT_CAPABILITIES from ipalib import api # The XMLRPC client is in "six.moves.xmlrpc_client", but pylint # cannot handle that try: from xmlrpclib import (Binary, Fault, DateTime, dumps, loads, ServerProxy, Transport, ProtocolError, MININT, MAXINT) except ImportError: # pylint: disable=import-error from xmlrpc.client import (Binary, Fault, DateTime, dumps, loads, ServerProxy, Transport, ProtocolError, MININT, MAXINT) # pylint: disable=import-error if six.PY3: from http.client import RemoteDisconnected else: from httplib import BadStatusLine as RemoteDisconnected # pylint: enable=import-error if six.PY3: unicode = str COOKIE_NAME = 'ipa_session' CCACHE_COOKIE_KEY = 'X-IPA-Session-Cookie' errors_by_code = dict((e.errno, e) for e in public_errors) def update_persistent_client_session_data(principal, data): ''' Given a principal create or update the session data for that principal in the persistent secure storage. Raises ValueError if unable to perform the action for any reason. ''' try: session_storage.store_data(principal, CCACHE_COOKIE_KEY, data) except Exception as e: raise ValueError(str(e)) def read_persistent_client_session_data(principal): ''' Given a principal return the stored session data for that principal from the persistent secure storage. Raises ValueError if unable to perform the action for any reason. ''' try: return session_storage.get_data(principal, CCACHE_COOKIE_KEY) except Exception as e: raise ValueError(str(e)) def delete_persistent_client_session_data(principal): ''' Given a principal remove the session data for that principal from the persistent secure storage. Raises ValueError if unable to perform the action for any reason. ''' try: session_storage.remove_data(principal, CCACHE_COOKIE_KEY) except Exception as e: raise ValueError(str(e)) def xml_wrap(value, version): """ Wrap all ``str`` in ``xmlrpc.client.Binary``. Because ``xmlrpc.client.dumps()`` will itself convert all ``unicode`` instances into UTF-8 encoded ``str`` instances, we don't do it here. So in total, when encoding data for an XML-RPC packet, the following transformations occur: * All ``str`` instances are treated as binary data and are wrapped in an ``xmlrpc.client.Binary()`` instance. * Only ``unicode`` instances are treated as character data. They get converted to UTF-8 encoded ``str`` instances (although as mentioned, not by this function). Also see `xml_unwrap()`. :param value: The simple scalar or simple compound value to wrap. """ if type(value) in (list, tuple): return tuple(xml_wrap(v, version) for v in value) if isinstance(value, dict): return dict( (k, xml_wrap(v, version)) for (k, v) in value.items() ) if type(value) is bytes: return Binary(value) if type(value) is Decimal: # transfer Decimal as a string return unicode(value) if isinstance(value, six.integer_types) and (value < MININT or value > MAXINT): return unicode(value) if isinstance(value, DN): return str(value) # Encode datetime.datetime objects as xmlrpc.client.DateTime objects if isinstance(value, datetime.datetime): if capabilities.client_has_capability(version, 'datetime_values'): return DateTime(value) else: return value.strftime(LDAP_GENERALIZED_TIME_FORMAT) if isinstance(value, DNSName): if capabilities.client_has_capability(version, 'dns_name_values'): return {'__dns_name__': unicode(value)} else: return unicode(value) if isinstance(value, Principal): return unicode(value) assert type(value) in (unicode, float, bool, type(None)) + six.integer_types return value def xml_unwrap(value, encoding='UTF-8'): """ Unwrap all ``xmlrpc.Binary``, decode all ``str`` into ``unicode``. When decoding data from an XML-RPC packet, the following transformations occur: * The binary payloads of all ``xmlrpc.client.Binary`` instances are returned as ``str`` instances. * All ``str`` instances are treated as UTF-8 encoded Unicode strings. They are decoded and the resulting ``unicode`` instance is returned. Also see `xml_wrap()`. :param value: The value to unwrap. :param encoding: The Unicode encoding to use (defaults to ``'UTF-8'``). """ if type(value) in (list, tuple): return tuple(xml_unwrap(v, encoding) for v in value) if type(value) is dict: if '__dns_name__' in value: return DNSName(value['__dns_name__']) else: return dict( (k, xml_unwrap(v, encoding)) for (k, v) in value.items() ) if isinstance(value, bytes): return value.decode(encoding) if isinstance(value, Binary): assert type(value.data) is bytes return value.data if isinstance(value, DateTime): # xmlprc DateTime is converted to string of %Y%m%dT%H:%M:%S format return datetime.datetime.strptime(str(value), "%Y%m%dT%H:%M:%S") assert type(value) in (unicode, int, float, bool, type(None)) return value def xml_dumps(params, version, methodname=None, methodresponse=False, encoding='UTF-8'): """ Encode an XML-RPC data packet, transparently wraping ``params``. This function will wrap ``params`` using `xml_wrap()` and will then encode the XML-RPC data packet using ``xmlrpc.client.dumps()`` (from the Python standard library). For documentation on the ``xmlrpc.client.dumps()`` function, see: http://docs.python.org/library/xmlrpc.client.html#convenience-functions Also see `xml_loads()`. :param params: A ``tuple`` or an ``xmlrpc.client.Fault`` instance. :param methodname: The name of the method to call if this is a request. :param methodresponse: Set this to ``True`` if this is a response. :param encoding: The Unicode encoding to use (defaults to ``'UTF-8'``). """ if type(params) is tuple: params = xml_wrap(params, version) else: assert isinstance(params, Fault) return dumps(params, methodname=methodname, methodresponse=methodresponse, encoding=encoding, allow_none=True, ) class _JSONPrimer(dict): """Fast JSON primer and pre-converter Prepare a data structure for JSON serialization. In an ideal world, priming could be handled by the default hook of json.dumps(). Unfortunately the hook treats Python 2 str as text while FreeIPA considers str as bytes. The primer uses a couple of tricks to archive maximum performance: * O(1) type look instead of O(n) chain of costly isinstance() calls * __missing__ and __mro__ with caching to handle subclasses * inline code with minor code duplication (func lookup in enc_list/dict) * avoid surplus function calls (e.g. func is _identity, obj.__class__ instead if type(obj)) * function default arguments to turn global into local lookups * avoid re-creation of bound method objects (e.g. result.append) * on-demand lookup of client capabilities with cached values Depending on the client version number, the primer converts: * bytes -> {'__base64__': b64encode} * datetime -> {'__datetime__': LDAP_GENERALIZED_TIME} * DNSName -> {'__dns_name__': unicode} The _ipa_obj_hook() functions unserializes the marked JSON objects to bytes, datetime and DNSName. :see: _ipa_obj_hook """ __slots__ = ('version', '_cap_datetime', '_cap_dnsname') _identity = object() def __init__(self, version, _identity=_identity): super(_JSONPrimer, self).__init__() self.version = version self._cap_datetime = None self._cap_dnsname = None self.update({ unicode: _identity, bool: _identity, type(None): _identity, float: _identity, Decimal: unicode, DN: str, Principal: unicode, DNSName: self._enc_dnsname, datetime.datetime: self._enc_datetime, bytes: self._enc_bytes, list: self._enc_list, tuple: self._enc_list, dict: self._enc_dict, }) # int, long for t in six.integer_types: self[t] = _identity def __missing__(self, typ): # walk MRO to find best match for c in typ.__mro__: if c in self: self[typ] = self[c] return self[c] # use issubclass to check for registered ABCs for c in self: if issubclass(typ, c): self[typ] = self[c] return self[c] raise TypeError(typ) def convert(self, obj, _identity=_identity): # obj.__class__ is twice as fast as type(obj) func = self[obj.__class__] return obj if func is _identity else func(obj) def _enc_datetime(self, val): cap = self._cap_datetime if cap is None: cap = capabilities.client_has_capability(self.version, 'datetime_values') self._cap_datetime = cap if cap: return {'__datetime__': val.strftime(LDAP_GENERALIZED_TIME_FORMAT)} else: return val.strftime(LDAP_GENERALIZED_TIME_FORMAT) def _enc_dnsname(self, val): cap = self._cap_dnsname if cap is None: cap = capabilities.client_has_capability(self.version, 'dns_name_values') self._cap_dnsname = cap if cap: return {'__dns_name__': unicode(val)} else: return unicode(val) def _enc_bytes(self, val): encoded = base64.b64encode(val) if not six.PY2: encoded = encoded.decode('ascii') return {'__base64__': encoded} def _enc_list(self, val, _identity=_identity): result = [] append = result.append for v in val: func = self[v.__class__] append(v if func is _identity else func(v)) return result def _enc_dict(self, val, _identity=_identity, _iteritems=six.iteritems): result = {} for k, v in _iteritems(val): func = self[v.__class__] result[k] = v if func is _identity else func(v) return result def json_encode_binary(val, version, pretty_print=False): """Serialize a Python object structure to JSON :param object val: Python object structure :param str version: client version :param bool pretty_print: indent and sort JSON (warning: slow!) :return: text :note: pretty printing triggers a slow path in Python's JSON module. Only use pretty_print in debug mode. """ result = _JSONPrimer(version).convert(val) if pretty_print: return json.dumps(result, indent=4, sort_keys=True) else: return json.dumps(result) def _ipa_obj_hook(dct, _iteritems=six.iteritems, _list=list): """JSON object hook :see: _JSONPrimer """ if '__base64__' in dct: return base64.b64decode(dct['__base64__']) elif '__datetime__' in dct: return datetime.datetime.strptime(dct['__datetime__'], LDAP_GENERALIZED_TIME_FORMAT) elif '__dns_name__' in dct: return DNSName(dct['__dns_name__']) else: # XXX tests assume tuples. Is this really necessary? for k, v in _iteritems(dct): if v.__class__ is _list: dct[k] = tuple(v) return dct def json_decode_binary(val):


context:
space:
mode: