| 38 | |
| 39 | |
| 40 | class AttributeService(object): |
| 41 | |
| 42 | def __init__(self, aspace): |
| 43 | self.logger = logging.getLogger(__name__) |
| 44 | self._aspace = aspace |
| 45 | |
| 46 | def read(self, params): |
| 47 | self.logger.debug("read %s", params) |
| 48 | res = [] |
| 49 | for readvalue in params.NodesToRead: |
| 50 | res.append(self._aspace.get_attribute_value(readvalue.NodeId, readvalue.AttributeId)) |
| 51 | return res |
| 52 | |
| 53 | def write(self, params, user=UserManager.User.Admin): |
| 54 | self.logger.debug("write %s as user %s", params, user) |
| 55 | res = [] |
| 56 | for writevalue in params.NodesToWrite: |
| 57 | if user != UserManager.User.Admin: |
| 58 | if writevalue.AttributeId != ua.AttributeIds.Value: |
| 59 | res.append(ua.StatusCode(ua.StatusCodes.BadUserAccessDenied)) |
| 60 | continue |
| 61 | al = self._aspace.get_attribute_value(writevalue.NodeId, ua.AttributeIds.AccessLevel) |
| 62 | ual = self._aspace.get_attribute_value(writevalue.NodeId, ua.AttributeIds.UserAccessLevel) |
| 63 | if not ua.ua_binary.test_bit(al.Value.Value, ua.AccessLevel.CurrentWrite) or not ua.ua_binary.test_bit(ual.Value.Value, ua.AccessLevel.CurrentWrite): |
| 64 | res.append(ua.StatusCode(ua.StatusCodes.BadUserAccessDenied)) |
| 65 | continue |
| 66 | res.append(self._aspace.set_attribute_value(writevalue.NodeId, writevalue.AttributeId, writevalue.Value)) |
| 67 | return res |
| 68 | |
| 69 | |
| 70 | class ViewService(object): |