409 lines
13 KiB
Python
409 lines
13 KiB
Python
'''
|
|
VERIDIC - Towards a centralized access control system
|
|
|
|
Copyright (C) 2011 Mikael Ates
|
|
|
|
This program is free software: you can redistribute it and/or modify
|
|
it under the terms of the GNU Affero 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 Affero General Public License for more details.
|
|
|
|
You should have received a copy of the GNU Affero General Public License
|
|
along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
'''
|
|
|
|
import re
|
|
import random
|
|
import string
|
|
|
|
from django.db import models
|
|
from django.contrib.auth.models import User
|
|
from django.utils.translation import ugettext as _
|
|
|
|
from acs.xacml.constants import *
|
|
|
|
SOURCE_TYPE = (
|
|
('DIRECT', _('Direct trusted source')),
|
|
('ANCHOR', _('Trust anchor')))
|
|
|
|
|
|
class Source(models.Model):
|
|
name = models.CharField(max_length = 100, unique=True)
|
|
public_key_or_ssl_certificate = models.TextField(blank=True)
|
|
type_source = models.CharField(
|
|
max_length = 60, choices = SOURCE_TYPE,
|
|
verbose_name = '',
|
|
default = 'DIRECT')
|
|
|
|
def __unicode__(self):
|
|
return 'Source %s of type %s' % (self.name, self.type_source)
|
|
|
|
|
|
class AttributeDefinition(models.Model):
|
|
attribute_name = models.CharField(max_length = 100, unique=True)
|
|
attribute_type = models.CharField(
|
|
max_length = 100, choices = XACML_DATA_TYPE,
|
|
verbose_name = '',
|
|
default = ACS_XACML_DATATYPE_STRING)
|
|
|
|
def __unicode__(self):
|
|
return self.attribute_name
|
|
|
|
|
|
class AttributeNamespace(models.Model):
|
|
friendly_name = models.CharField(max_length = 100, unique=True)
|
|
identifier = models.CharField(max_length = 100, unique=True)
|
|
|
|
|
|
class AttributeMapInNamespace(models.Model):
|
|
namespace = models.ForeignKey(AttributeNamespace)
|
|
attribute_definition = models.ForeignKey(AttributeDefinition)
|
|
|
|
|
|
class AttributeName(models.Model):
|
|
attribute_map = models.ForeignKey(AttributeMapInNamespace)
|
|
name = models.CharField(max_length = 100)
|
|
|
|
|
|
class AttributeData(models.Model):
|
|
definition = models.ForeignKey(AttributeDefinition)
|
|
|
|
|
|
class StringM(models.Model):
|
|
'''http://www.w3.org/2001/XMLSchema#string'''
|
|
data = models.ForeignKey(AttributeData)
|
|
value = models.CharField(max_length = 100)
|
|
|
|
def __unicode__(self):
|
|
return self.value
|
|
|
|
|
|
class BooleanM(models.Model):
|
|
'''http://www.w3.org/2001/XMLSchema#boolean'''
|
|
data = models.ForeignKey(AttributeData)
|
|
value = models.BooleanField()
|
|
|
|
def __unicode__(self):
|
|
return str(self.value)
|
|
|
|
|
|
class IntegerM(models.Model):
|
|
'''http://www.w3.org/2001/XMLSchema#integer'''
|
|
data = models.ForeignKey(AttributeData)
|
|
value = models.IntegerField()
|
|
|
|
def __unicode__(self):
|
|
return str(self.value)
|
|
|
|
|
|
class DoubleM(models.Model):
|
|
'''http://www.w3.org/2001/XMLSchema#double'''
|
|
data = models.ForeignKey(AttributeData)
|
|
value = models.FloatField()
|
|
|
|
def __unicode__(self):
|
|
return str(self.value)
|
|
|
|
|
|
class TimeM(models.Model):
|
|
'''http://www.w3.org/2001/XMLSchema#time'''
|
|
data = models.ForeignKey(AttributeData)
|
|
value = models.TimeField()
|
|
|
|
def __unicode__(self):
|
|
return str(self.value)
|
|
|
|
|
|
class DateM(models.Model):
|
|
'''http://www.w3.org/2001/XMLSchema#date'''
|
|
data = models.ForeignKey(AttributeData)
|
|
value = models.DateField()
|
|
|
|
def __unicode__(self):
|
|
return str(self.value)
|
|
|
|
|
|
class DateTimeM(models.Model):
|
|
'''http://www.w3.org/2001/XMLSchema#dateTime'''
|
|
data = models.ForeignKey(AttributeData)
|
|
value = models.DateTimeField()
|
|
|
|
def __unicode__(self):
|
|
return str(self.value)
|
|
|
|
|
|
class Rfc822NameM(models.Model):
|
|
'''urn:oasis:names:tc:xacml:1.0:data-type:rfc822Name'''
|
|
data = models.ForeignKey(AttributeData)
|
|
value = models.EmailField()
|
|
|
|
def __unicode__(self):
|
|
return str(self.value)
|
|
|
|
|
|
class IpAddressM(models.Model):
|
|
'''urn:oasis:names:tc:xacml:2.0:data-type:ipAddress'''
|
|
data = models.ForeignKey(AttributeData)
|
|
value = models.IPAddressField()
|
|
|
|
def __unicode__(self):
|
|
return str(self.value)
|
|
|
|
|
|
class Certificate(models.Model):
|
|
raw_data = models.TextField()
|
|
|
|
|
|
class AssertionAny(models.Model):
|
|
|
|
def get_assertion_instance(self):
|
|
try:
|
|
return self.assertiondefinition
|
|
except:
|
|
pass
|
|
try:
|
|
return self.assertiondata
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
|
|
class AssertionDefinition(AssertionAny):
|
|
attribute_definition = models.ForeignKey(AttributeDefinition)
|
|
|
|
def __unicode__(self):
|
|
sources = AttachedSource.objects.filter(assertion=self)
|
|
return "attribute %s from %s" \
|
|
% (str(self.attribute_definition),
|
|
[str(x.source) for x in sources])
|
|
|
|
|
|
class AttachedSource(models.Model):
|
|
source = models.ForeignKey(Source)
|
|
assertion = models.ForeignKey(AssertionDefinition,
|
|
related_name = 'assertion')
|
|
|
|
def __unicode__(self):
|
|
return "Source %s attached to %s" \
|
|
% (str(self.source), str(self.assertion))
|
|
|
|
|
|
class AssertionData(AssertionAny):
|
|
'''
|
|
An assertion data may have no source or a unique source attached.
|
|
No source when used in a rule for a comparison for instance.
|
|
A unique source when used in a profile for instance.
|
|
'''
|
|
attribute_data = models.ForeignKey(AttributeData)
|
|
source = models.ForeignKey(Source, null=True, blank=True)
|
|
certificate = models.ForeignKey(Certificate, null=True, blank=True)
|
|
creation_date = models.DateTimeField(auto_now_add=True)
|
|
expiration_date = models.DateTimeField(null=True, blank=True)
|
|
|
|
def get_values(self):
|
|
values = None
|
|
definition = self.attribute_data.definition
|
|
if definition.attribute_type == ACS_XACML_DATATYPE_STRING:
|
|
values = StringM.objects.filter(data=self.attribute_data)
|
|
elif definition.attribute_type == ACS_XACML_DATATYPE_BOOLEAN:
|
|
values = BooleanM.objects.filter(data=self.attribute_data)
|
|
elif definition.attribute_type == ACS_XACML_DATATYPE_INTEGER:
|
|
values = IntegerM.objects.filter(data=self.attribute_data)
|
|
elif definition.attribute_type == ACS_XACML_DATATYPE_DOUBLE:
|
|
values = DoubleM.objects.filter(data=self.attribute_data)
|
|
elif definition.attribute_type == ACS_XACML_DATATYPE_TIME:
|
|
values = TimeM.objects.filter(data=self.attribute_data)
|
|
elif definition.attribute_type == ACS_XACML_DATATYPE_DATE:
|
|
values = DateM.objects.filter(data=self.attribute_data)
|
|
elif definition.attribute_type == ACS_XACML_DATATYPE_DATETIME:
|
|
values = DateTimeM.objects.filter(data=self.attribute_data)
|
|
elif definition.attribute_type == ACS_XACML_DATATYPE_RFC822NAME:
|
|
values = Rfc822NameM.objects.filter(data=self.attribute_data)
|
|
elif definition.attribute_type == ACS_XACML_DATATYPE_IPADDRESS:
|
|
values = IpAddressM.objects.filter(data=self.attribute_data)
|
|
return values
|
|
|
|
def __unicode__(self):
|
|
values = self.get_values()
|
|
s = "Attribute %s with values %s" \
|
|
% (str(self.attribute_data.definition),
|
|
[str(x) for x in values])
|
|
if self.source:
|
|
s += ' (provided by %s)' % str(self.source)
|
|
if self.certificate:
|
|
s += ' (signed with %s)' % str(self.certificate)
|
|
if self.expiration_date:
|
|
s += ' (expires on %s)' % str(self.expiration_date)
|
|
return s
|
|
|
|
|
|
class UserAttributeProfile(models.Model):
|
|
user = models.OneToOneField(User, null=True, blank=True,
|
|
related_name='profile')
|
|
assertions = models.ManyToManyField(AssertionData,
|
|
verbose_name=_('data_assertions'), blank=True)
|
|
|
|
def __unicode__(self):
|
|
if not self.user:
|
|
s = 'Anonymous profile'
|
|
else:
|
|
s = 'Profile of user %s' % self.user
|
|
if not self.assertions:
|
|
return s + ' is empty.'
|
|
else:
|
|
for ad in self.assertions.all():
|
|
attribute_data = ad.attribute_data
|
|
s += " - assertion from %s with definition %s and values %s" \
|
|
% (ad.source, attribute_data.definition.id,
|
|
str([str(x.value) for x in ad.get_values()]))
|
|
return s
|
|
|
|
|
|
'''
|
|
An ABAC rule is a string containing logical statements (and, or, not) and
|
|
the identifiers of predicates.
|
|
|
|
To make feasible to use the the model instance identifiers of predicates,
|
|
all predicates objects have a unique identifiers with a model defined per
|
|
predicate type, a parent model is used: Predicate.
|
|
|
|
All predicate types (e.g. PredicateRequired) herit from Predicate.
|
|
|
|
The identifiers used are thus the identifiers of the instances of the
|
|
model Predicate
|
|
'''
|
|
|
|
|
|
class Predicate(models.Model):
|
|
rule = models.ForeignKey('AbacRule')
|
|
|
|
def get_predicate_instance(self):
|
|
try:
|
|
return self.predicaterole
|
|
except:
|
|
pass
|
|
try:
|
|
return self.predicaterequired
|
|
except:
|
|
pass
|
|
try:
|
|
return self.predicatecomparison
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
|
|
'''
|
|
To say that an attribute is required
|
|
'''
|
|
|
|
|
|
class PredicateRequired(Predicate):
|
|
definition = models.ForeignKey(AssertionDefinition)
|
|
single_value = models.BooleanField(default=False)
|
|
|
|
def __unicode__(self):
|
|
if self.single_value:
|
|
return "Predicate required: %s - \
|
|
The attribute must be single-valued" % str(self.definition)
|
|
return "Predicate required: %s" % str(self.definition)
|
|
|
|
|
|
class PredicateRole(Predicate):
|
|
'''
|
|
Role are only handled from the ACS role tree
|
|
|
|
Else, use attributes equality with roles provided as attributes from
|
|
sources.
|
|
'''
|
|
role = models.ForeignKey('acs.Role')
|
|
|
|
def __unicode__(self):
|
|
return "Predicate role on %s" % str(self.role)
|
|
|
|
|
|
class PredicateComparison(Predicate):
|
|
operand1 = models.ForeignKey(AssertionAny, related_name = 'operand1')
|
|
operand2 = models.ForeignKey(AssertionAny, related_name = 'operand2')
|
|
operand1_single_value = models.BooleanField(default=False)
|
|
operand2_single_value = models.BooleanField(default=False)
|
|
comparison_type = models.CharField(max_length = 100,
|
|
choices = XACML_COMPARISON_TYPE,
|
|
verbose_name = 'type of comparison',
|
|
default = ACS_XACML_COMPARISON_EQUALITY_STRING)
|
|
multivalues = models.CharField(max_length = 100,
|
|
verbose_name = 'How to handle multivalued attributes',
|
|
default = 'NO_MULTIVALUES')
|
|
multivalues_explanation = models.CharField(max_length = 500, blank=True)
|
|
|
|
def __unicode__(self):
|
|
operator = ''
|
|
if self.comparison_type in XACML_COMPARISON_EQUALITY:
|
|
operator = '='
|
|
elif self.comparison_type in ACS_XACML_COMPARISON_LT:
|
|
operator = '<'
|
|
elif self.comparison_type in ACS_XACML_COMPARISON_LT_OE:
|
|
operator = '<='
|
|
elif self.comparison_type in ACS_XACML_COMPARISON_GRT:
|
|
operator = '>'
|
|
elif self.comparison_type in ACS_XACML_COMPARISON_GRT_OE:
|
|
operator = '>='
|
|
s = 'Predicate comparison: %s %s %s (' \
|
|
% (str(self.operand1.get_assertion_instance()),
|
|
operator, str(self.operand2.get_assertion_instance()),
|
|
)
|
|
if self.operand1_single_value:
|
|
s += 'operand one requires a single-valued attribute - '
|
|
if self.operand2_single_value:
|
|
s += 'operand two requires a single-valued attribute - '
|
|
if not self.operand1_single_value or not self.operand2_single_value:
|
|
if not self.multivalues_explanation:
|
|
s += 'multivalues management is %s' \
|
|
% self.multivalues
|
|
else:
|
|
s += 'The multivalues management is as follows: %s' \
|
|
% self.multivalues_explanation
|
|
s += ')'
|
|
return s
|
|
|
|
|
|
'''
|
|
Predicate are str(id) of attribute definitions
|
|
'''
|
|
|
|
|
|
class AbacRule(models.Model):
|
|
expression = models.CharField(max_length = 2048)
|
|
|
|
def __unicode__(self):
|
|
predicates1 = {}
|
|
predicates2 = {}
|
|
expression = self.expression
|
|
'''
|
|
Here we substitute primary keys with the predicate description.
|
|
|
|
The substitution might fail if some descriptions contains
|
|
primary keys since with an iterative substitution, a
|
|
primary key resulting from a previsous substitution would be
|
|
replaced by another description.
|
|
|
|
To prevent this we realize a two-round substitution with unique
|
|
identifiers.
|
|
'''
|
|
for p in Predicate.objects.filter(rule=self):
|
|
rdm_str = ''.join(\
|
|
random.choice(string.ascii_uppercase) for x in range(8))
|
|
predicates1[p.id] = rdm_str
|
|
predicates2[rdm_str] = str(p.get_predicate_instance())
|
|
for key in predicates1.keys():
|
|
expression = re.sub(str(key), str(predicates1[key]),
|
|
expression)
|
|
for key in predicates2.keys():
|
|
expression = re.sub(str(key), str(predicates2[key]), expression)
|
|
return expression
|