/usr/share/pyshared/nwss/base.py is in python-nwsserver 2.0.0-2build1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 140 141 142 143 144 145 146 147 148 149 150 151 152 | #
# Copyright (c) 2005-2009, REvolution Computing, Inc.
#
# NetWorkSpaces 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 2 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, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
# USA
#
"""
Core NetWorkSpaces server - common bits.
"""
import os, mmap, traceback
from twisted.python import log
# bit codings for the descriptor.
DIRECT_STRING = 1
class ServerException(Exception):
"""Base class for all exceptions raised by this module."""
def __str__(self):
#pylint: disable-msg=W0141
return '%s[%s]' % (self.__class__.__name__,
' '.join(map(str, self.args)))
def __repr__(self):
#pylint: disable-msg=W0141
return '%s(%s)' % (self.__class__.__name__,
', '.join(map(repr, self.args)))
class WorkspaceFailure(ServerException):
"""Exception thrown by Workspace to signal that a normal (i.e. user-level)
error has occurred during the operation.
"""
def __init__(self, msg, status=1):
ServerException.__init__(self, msg)
self.status = status
class BadModeException(ServerException):
"""Variable cannot be set to specified mode."""
class NoSuchVariableException(ServerException):
"""No variable by specified name."""
class Response(object):
#pylint: disable-msg=R0903
"""Response from the server to the client."""
def __init__(self, metadata=None, value=None):
self.status = 0
if metadata is None:
metadata = {}
self.metadata = metadata
self.value = value
self.iterstate = None
class Value(object):
"""Value wrapper class handling out-of-band transmission of long data."""
def __init__(self, desc, val):
"""Initialize a value object.
Arguments:
desc - type descriptor for object
val - either a string or a (filename, length) tuple
"""
self.__type_descriptor = desc
self._val = val
self._consumed = False
if isinstance(val, str):
self._long = False
self._length = len(val)
else:
# if it's not a string, assume it's a tuple: (filename, length)
self._long = True
self._length = val[1]
def consumed(self):
"""Flag this value as consumed.
"""
self._consumed = True
def access_complete(self):
"""Notify this value that it has been sent to the client, and should
now deallocate its file if it has been consumed.
"""
if self._consumed:
self.close()
def close(self):
"""Deallocate any resources associated with this value."""
if self._long:
try:
os.remove(self._val[0])
except OSError:
log.msg('error removing file %s' % self._val[0])
traceback.print_exc()
def get_file(self):
"""Memory map the file associated with this long value. If this is not
a long value, this method will fail.
"""
assert self._long, 'get_file illegally called on string value'
datafile = open(self._val[0], 'rb')
memory = mmap.mmap(datafile.fileno(), self._length,
access=mmap.ACCESS_READ)
datafile.close()
return memory
def is_large(self):
"""Is this a large value?"""
return self._long
def __get_type_descriptor(self):
"""Get the type descriptor for this value."""
return self.__type_descriptor
type_descriptor = property(__get_type_descriptor)
def val(self):
"""Get the raw value for this non-long value.
If this is a long value, this method will fail."""
assert not self._long, 'val illegally called on long value'
return self._val
def set_val(self, data):
"""Set the raw value for this non-long value.
If this is a long value, this method will fail."""
assert not self._long, 'val illegally called on long value'
self._val = data
def length(self):
"""Get the length of this value in bytes."""
return self._length
ERROR_VALUE = Value(0, '')
|