entrouvert
/
pyoidc-ozwillo
Archived
17
0
Fork 0
Code Issues Pull Requests Releases Activity

Updated parameter names

This commit is contained in:
Roland Hedberg 2013-01-01 16:51:52 +01:00
parent 6c4a7788c7
commit 85a82c27b0
23 changed files with 511 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.idea/libraries/sass_stdlib.xml Normal file
View File

@ -0,0 +1,8 @@
<component name="libraryTable">
<library name="sass-stdlib">
<CLASSES />
<SOURCES>
<root url="file://$APPLICATION_HOME_DIR$/plugins/sass/lib/stubs/sass_functions.scss" />
</SOURCES>
</library>
</component>

2
src/oic/oauth2/provider.py
View File

@ -133,7 +133,7 @@ class Provider(object):
except KeyError:
return BadRequest("Unknown session identifier")
_sdb.update(sid, "user_id", dic["login"][0])
_sdb.update(sid, "sub", dic["login"][0])
LOG_DEBUG("session[\"authzreq\"] = %s" % session["authzreq"])
#_log_info( "type: %s" % type(session["authzreq"]))

6
src/oic/oic/__init__.py
View File

@ -304,7 +304,7 @@ PREFERENCE2PROVIDER = {
"id_token_encrypted_response_alg": "id_token_encryption_alg_values_supported",
"id_token_encrypted_response_enc": "id_token_encryption_enc_values_supported",
"default_acr": "acrs_supported",
"user_id_type": "user_id_types_supported",
"subbject_type": "subbject_types_supported",
"token_endpoint_auth_alg": "token_endpoint_auth_algs_supported",
#"request_object_signing_alg": "request_object_signing_alg_values_supported
}
@ -1243,7 +1243,7 @@ class Server(oauth2.Server):
_args = user_info.to_dict()
# Make sure that there are no name clashes
for key in ["iss", "user_id", "aud", "exp", "acr", "nonce",
for key in ["iss", "sub", "aud", "exp", "acr", "nonce",
"auth_time"]:
try:
del _args[key]
@ -1257,7 +1257,7 @@ class Server(oauth2.Server):
if access_token:
_args["at_hash"] = jws.left_hash(access_token, halg)
idt = IdToken(iss=issuer, user_id=session["user_id"],
idt = IdToken(iss=issuer, sub=session["sub"],
aud = session["client_id"],
exp = time_util.epoch_in_a_while(**inawhile), acr=loa,
iat = time_util.utc_now(),

8
src/oic/oic/claims_provider.py
View File

@ -31,7 +31,7 @@ from oic.utils.http_util import Response, Unauthorized
logger = logging.getLogger(__name__)
class UserClaimsRequest(Message):
c_param = {"user_id": SINGLE_REQUIRED_STRING,
c_param = {"sub": SINGLE_REQUIRED_STRING,
"client_id": SINGLE_REQUIRED_STRING,
"client_secret": SINGLE_REQUIRED_STRING,
"claims_names": REQUIRED_LIST_OF_STRINGS}
@ -143,14 +143,14 @@ class ClaimsServer(Provider):
_log_info("User info claims: %s" % uic)
#oicsrv, userdb, user_id, client_id="", user_info_claims=None
info = self.function["userinfo"](self, self.userdb, ucreq["user_id"],
#oicsrv, userdb, subject, client_id="", user_info_claims=None
info = self.function["userinfo"](self, self.userdb, ucreq["sub"],
ucreq["client_id"],
user_info_claims=uic)
_log_info("User info: %s" % info.to_dict())
if self.do_aggregation(info, ucreq["user_id"]):
if self.do_aggregation(info, ucreq["sub"]):
cresp = self._aggregation(info)
else:
cresp = self._distributed(info)

2
src/oic/oic/consumer.py
View File

@ -137,7 +137,7 @@ IGNORE = ["request2endpoint", "response2error", "grant_class", "token_class"]
CONSUMER_PREF_ARGS =[
"token_endpoint_auth_type",
"user_id_type",
"subject_type",
"require_signed_request_object",
"userinfo_signed_response_algs",
"userinfo_encrypted_response_alg",

47
src/oic/oic/provider.py
View File

@ -242,6 +242,9 @@ class Provider(AProvider):
"""
MUST NOT contain a fragment
MAY contain query component
:return: An error response if the redirect URI is faulty otherwise
None
"""
try:
_redirect_uri = urlparse.unquote(areq["redirect_uri"])
@ -348,7 +351,7 @@ class Provider(AProvider):
sid = client_info["sector_id"]
except KeyError:
try:
uit = client_info["user_id_type"]
uit = client_info["subject_type"]
if uit == "pairwise":
sid = _base
except KeyError:
@ -449,11 +452,11 @@ class Provider(AProvider):
if openid_req:
try:
user = openid_req["id_token"]["claims"]["user_id"]["value"]
user = openid_req["id_token"]["claims"]["sub"]["value"]
except KeyError:
user = ""
elif "id_token" in areq:
user = areq["id_token"]["user_id"]
user = areq["id_token"]["sub"]
else:
user = ""
@ -496,9 +499,9 @@ class Provider(AProvider):
cookie = None
if self.authn_as:
user_id = self.authn_as
_log_debug("Implicit authenticated as %s" % user_id)
_sdb.update(sid, "local_user_id", user_id)
sub = self.authn_as
_log_debug("Implicit authenticated as %s" % sub)
_sdb.update(sid, "local_sub", sub)
(redirect_uri, reply) = self.get_redirect_uri(areq)
client_info = self.cdb[areq["client_id"]]
sector_id = self.get_sector_id(redirect_uri, client_info)
@ -508,11 +511,11 @@ class Provider(AProvider):
except KeyError:
preferred_id_type = self.preferred_id_type
self.sdb.do_userid(sid, user_id, sector_id, preferred_id_type)
self.sdb.do_userid(sid, sub, sector_id, preferred_id_type)
_log_debug("session: %s" % _sdb[sid])
_log_debug("uid2sid: %s" % _sdb.uid2sid)
return self.authenticated(environ, start_response, active_auth=sid,
areq=areq, user=user_id)
areq=areq, user=sub)
if openid_req:
_log_info("Request: %s" % openid_req.to_dict())
@ -534,8 +537,8 @@ class Provider(AProvider):
_log_info("- SSO -")
_scode = base64.b64decode(key)
_log_debug("OLD session: %s" % _sdb[_scode])
user = self.sdb[_scode]["user_id"]
_sdb.update(sid, "user_id", user)
user = self.sdb[_scode]["sub"]
_sdb.update(sid, "sub", user)
return self.authenticated(environ,
start_response,
active_auth=_scode,
@ -555,8 +558,8 @@ class Provider(AProvider):
try:
_log_info("- SSO -")
_scode = base64.b64decode(key)
user = self.sdb[_scode]["user_id"]
_sdb.update(sid, "user_id", user)
user = self.sdb[_scode]["sub"]
_sdb.update(sid, "sub", user)
# This happens if a valid cookie is presented
return self.authenticated(environ, start_response,
active_auth=_scode,
@ -866,11 +869,11 @@ class Provider(AProvider):
logger.debug("userdb: %s" % self.userdb.keys())
logger.debug("Session info: %s" % session)
info = self.function["userinfo"](self, self.userdb,
session["local_user_id"],
session["local_sub"],
session["client_id"],
userinfo_claims)
info["user_id"] = session["user_id"]
info["sub"] = session["sub"]
logger.debug("user_info_response: %s" % (info,))
return info
@ -1241,7 +1244,7 @@ class Provider(AProvider):
"code id_token",
"token id_token",
"code token id_token"],
user_id_types_supported=["public", "pairwise"],
subject_types_supported=["public", "pairwise"],
#request_object_algs_supported=["HS256"]
)
@ -1351,7 +1354,7 @@ class Provider(AProvider):
issue_new_code = False
if "active_auth" in kwargs:
scode = kwargs["active_auth"]
user_id = kwargs["user"]
sub = kwargs["user"]
areq = kwargs["areq"]
client_info = self.cdb[areq["client_id"]]
if "code" in areq["response_type"]:
@ -1368,7 +1371,7 @@ class Provider(AProvider):
pass
try:
(verified, user_id) = self.function["verify_user"](dic)
(verified, sub) = self.function["verify_user"](dic)
if not verified:
resp = Unauthorized("Wrong password")
return resp(environ, start_response)
@ -1376,7 +1379,7 @@ class Provider(AProvider):
resp = Unauthorized("%s" % (err,))
return resp(environ, start_response)
_log_debug("verified user_id: %s" % user_id)
_log_debug("verified sub: %s" % sub)
try:
# Use the session identifier to find the session information
@ -1398,20 +1401,20 @@ class Provider(AProvider):
areq = AuthorizationRequest().deserialize(asession["authzreq"],
"json")
self.sdb.update(scode, "local_user_id", user_id)
self.sdb.update(scode, "local_sub", sub)
(redirect_uri, reply) = self.get_redirect_uri(areq)
client_info = self.cdb[areq["client_id"]]
sector_id = self.get_sector_id(redirect_uri, client_info)
try:
preferred_id_type = client_info["user_id_type"]
preferred_id_type = client_info["subject_type"]
except KeyError:
preferred_id_type = self.preferred_id_type
_log_debug("sector_id: %s, preferred_id_type: %s" % (sector_id,
preferred_id_type))
self.sdb.do_userid(scode, user_id, sector_id, preferred_id_type)
self.sdb.do_userid(scode, sub, sector_id, preferred_id_type)
_log_debug("areq: %s" % areq)
_log_debug("session: %s" % self.sdb[scode])
@ -1419,7 +1422,7 @@ class Provider(AProvider):
# Do the authorization
try:
permission = self.function["authorize"](user_id)
permission = self.function["authorize"](sub)
self.sdb.update(scode, "permission", permission)
except Exception:
raise

2
src/oic/utils/keyio.py
View File

@ -277,7 +277,7 @@ class KeyBundle(object):
pass
for key,val in self._key.items():
if val is []:
if val == []:
del self._key[key]
def __str__(self):

32
src/oic/utils/sdb.py
View File

@ -23,8 +23,8 @@ class ExpiredToken(Exception):
class WrongTokenType(Exception):
pass
def pairwise_id(user_id, sector_identifier, seed):
return hashlib.sha256("%s%s%s" % (user_id, sector_identifier, seed)).hexdigest()
def pairwise_id(sub, sector_identifier, seed):
return hashlib.sha256("%s%s%s" % (sub, sector_identifier, seed)).hexdigest()
class Crypt():
def __init__(self, password, mode=AES.MODE_CBC):
@ -163,13 +163,13 @@ class SessionDB(object):
(typ, key) = self.token.type_and_key(token)
return self.update(key, attribute, value)
def do_userid(self, sid, user_id, sector_id, preferred_id_type):
def do_userid(self, sid, sub, sector_id, preferred_id_type):
old = [""]
if preferred_id_type == "public":
uid = user_id
uid = sub
else:
uid = pairwise_id(user_id, sector_id, self.seed)
old.append(user_id)
uid = pairwise_id(sub, sector_id, self.seed)
old.append(sub)
logger.debug("uid: %s, old: %s" % (uid, old))
self.uid2sid[uid] = sid
@ -181,28 +181,28 @@ class SessionDB(object):
pass
logger.debug("uid2sid: %s" % self.uid2sid)
self._db[sid]["local_user_id"] = user_id
self._db[sid]["user_id"] = uid
self._db[sid]["local_sub"] = sub
self._db[sid]["sub"] = uid
return uid
def create_authz_session(self, user_id, areq, id_token=None, oidreq=None):
def create_authz_session(self, sub, areq, id_token=None, oidreq=None):
"""
:param user_id: Identifier for the user, this is the real identifier
:param sub: Identifier for the user, this is the real identifier
:param areq: The AuthorizationRequest instance
:param id_token: An IDToken instance
:param oidreq: An OpenIDRequest instance
:return: The session identifier, which is the database key
"""
sid = self.token.key(user=user_id, areq=areq)
sid = self.token.key(user=sub, areq=areq)
access_grant = self.token(sid=sid)
_dic = {
"oauth_state": "authz",
"local_user_id": user_id,
"user_id": user_id,
"local_sub": sub,
"sub": sub,
"code": access_grant,
"code_used": False,
"authzreq": areq.to_json(),
@ -232,7 +232,7 @@ class SessionDB(object):
_dic["oidreq"] = oidreq.to_json()
self._db[sid] = _dic
self.uid2sid[user_id] = sid
self.uid2sid[sub] = sid
return sid
def update_to_token(self, token=None, issue_refresh=True, id_token="",
@ -375,7 +375,7 @@ class SessionDB(object):
def duplicate(self, sinfo):
_dic = copy.copy(sinfo)
areq = AuthorizationRequest().from_json(_dic["authzreq"])
sid = self.token.key(user=_dic["user_id"], areq=areq)
sid = self.token.key(user=_dic["sub"], areq=areq)
_dic["code"] = self.token(sid=sid)
_dic["code_used"] = False
@ -389,5 +389,5 @@ class SessionDB(object):
pass
self._db[sid] = _dic
self.uid2sid[_dic["user_id"]] = sid
self.uid2sid[_dic["sub"]] = sid
return sid

50
tests/debug_1.py Normal file
View File

@ -0,0 +1,50 @@
import time
from oic.oic import Client
from oic.utils.time_util import utc_time_sans_frac
from oic.oic.message import IdToken
from oic.utils.keyio import KeyChain, KeyJar
__author__ = 'rohe0002'
CLIENT_SECRET = "abcdefghijklmnop"
CLIENT_ID = "client_1"
KC_HMAC_VS = KeyChain({"hmac": "abcdefghijklmnop"}, usage=["ver", "sig"])
KC_RSA = KeyChain(source="file://../oc3/certs/mycert.key", type="rsa",
usage=["ver", "sig"])
KC_HMAC_S = KeyChain({"hmac": "abcdefghijklmnop"}, usage=["sig"])
KEYJ = KeyJar()
KEYJ[""] = [KC_RSA, KC_HMAC_S]
KEYJ["client_1"] = [KC_HMAC_VS]
IDTOKEN = IdToken(iss="http://oic.example.org/", user_id="user_id",
aud=CLIENT_ID, exp=utc_time_sans_frac()+86400,
nonce="N0nce",
iat=time.time())
# ----------------- CLIENT --------------------
client = Client(CLIENT_ID)
client.redirect_uris = ["http://example.com/redirect"]
client.client_secret = CLIENT_SECRET
client.keyjar[""] = KC_RSA
claims = {
"name": {"essential": True},
"nickname": None,
"email": {"essential": True},
"verified": {"essential": True},
"picture": None
}
areq = client.construct_AuthorizationRequest(
request_args={"scope":"openid", "response_type":["code"]},
userinfo_claims={"claims":claims,
"preferred_locale":"en"},
idtoken_claims={"claims":{"auth_time": None,
"acr":{"values":["2"]}},
"max_age": 86400},
)
print areq

29
tests/debug_2.py Normal file
View File

@ -0,0 +1,29 @@
from oic.oic.consumer import Consumer
from oic.utils.keyio import KeyChain, KeyJar
__author__ = 'rohe0002'
from fakeoicsrv import MyFakeOICServer
CLIENT_SECRET = "abcdefghijklmnop"
CLIENT_ID = "client_1"
RSAPUB = "../oc3/certs/mycert.key"
KC_HMAC_VS = KeyChain({"hmac": CLIENT_SECRET}, usage=["ver", "sig"])
KC_RSA = KeyChain(source="file://%s" % RSAPUB, type="rsa", usage=["ver", "sig"])
KC_HMAC_S = KeyChain({"hmac": CLIENT_SECRET}, usage=["sig"])
SRVKEYS = KeyJar()
SRVKEYS[""] = [KC_RSA]
SRVKEYS["client_1"] = [KC_HMAC_VS, KC_RSA]
c = Consumer(None, None)
mfos = MyFakeOICServer("http://example.com")
mfos.keyjar = SRVKEYS
c.http_request = mfos.http_request
principal = "foo@example.com"
res = c.discover(principal)
info = c.provider_config(res)
assert info.type() == "ProviderConfigurationResponse"

96
tests/debug_cc.py Normal file
View File

@ -0,0 +1,96 @@
import StringIO
import sys
from oic.oic.message import OpenIDSchema
from oic.utils.keyio import KeyChain
from oic.oic.claims_provider import ClaimsClient
from oic.oic.claims_provider import ClaimsServer
#noinspection PyUnusedLocal
def user_info(oicsrv, userdb, user_id, client_id="", user_info_claims=None):
#print >> sys.stderr, "claims: %s" % user_info_claims
identity = userdb[user_id]
if user_info_claims:
result = {}
for key, restr in user_info_claims.items():
try:
result[key] = identity[key]
except KeyError:
if restr == {"essential": True}:
raise Exception("Missing property '%s'" % key)
else:
result = identity
return OpenIDSchema(**result)
class LOG():
def info(self, txt):
print >> sys.stdout, "INFO: %s" % txt
def error(self, txt):
print >> sys.stdout, "ERROR: %s" % txt
def debug(self, txt):
print >> sys.stdout, "DEBUG: %s" % txt
#noinspection PyUnusedLocal
def start_response(status, headers=None):
return
USERDB = {
"diana":{
"birthdate": "02/14/2012",
"gender": "female"
}
}
BASE_ENVIRON = {'SERVER_PROTOCOL': 'HTTP/1.1',
'REQUEST_METHOD': 'GET',
'QUERY_STRING': '',
'HTTP_CONNECTION': 'keep-alive',
'REMOTE_ADDR': '127.0.0.1',
'wsgi.url_scheme': 'http',
'SERVER_PORT': '8087',
'PATH_INFO': '/register',
'HTTP_HOST': 'localhost:8087',
'HTTP_ACCEPT': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'HTTP_ACCEPT_LANGUAGE': 'sv-se',
'CONTENT_TYPE': 'text/plain',
'REMOTE_HOST': '1.0.0.127.in-addr.arpa',
'HTTP_ACCEPT_ENCODING': 'gzip, deflate',
'COMMAND_MODE': 'unix2003'}
CDB = {
"client_1": { "client_secret": "hemlig"}
}
def verify_client(env, req, cdb):
return True
FUNCTIONS = {
"verify_client": verify_client,
"userinfo": user_info
}
cc = ClaimsClient(client_id="client_1")
cc.client_secret="hemlig"
req = cc.construct_UserClaimsRequest(request_args={"user_id": "diana",
"claims_names":["gender", "birthdate"]})
srv = ClaimsServer("name", None, CDB, FUNCTIONS, USERDB)
srv.keyjar[""] = [KeyChain(source="file://rsa.key", usage=["ver", "sig"])]
assert srv
environ = BASE_ENVIRON.copy()
environ["REQUEST_METHOD"] = "POST"
txt = req.to_urlencoded()
environ["CONTENT_LENGTH"] = len(txt)
fil = StringIO.StringIO(buf=txt)
environ["wsgi.input"] = fil
resp = srv.claims_endpoint(environ, start_response, LOG())
print resp

103
tests/debug_claims_srv2.py Normal file
View File

@ -0,0 +1,103 @@
import StringIO
import sys
from oic.oic.message import OpenIDSchema
from oic.utils.keyio import KeyChain
from oic.oic.claims_provider import UserClaimsResponse
from oic.oic.claims_provider import ClaimsClient
from oic.oic.claims_provider import ClaimsServer
__author__ = 'rohe0002'
def user_info(oicsrv, userdb, user_id, client_id="", user_info_claims=None):
#print >> sys.stderr, "claims: %s" % user_info_claims
identity = userdb[user_id]
if user_info_claims:
result = {}
for key, restr in user_info_claims["claims"].items():
try:
result[key] = identity[key]
except KeyError:
if restr == {"essential": True}:
raise Exception("Missing property '%s'" % key)
else:
result = identity
return OpenIDSchema(**result)
class LOG():
def info(self, txt):
print >> sys.stdout, "INFO: %s" % txt
def error(self, txt):
print >> sys.stdout, "ERROR: %s" % txt
def debug(self, txt):
print >> sys.stdout, "DEBUG: %s" % txt
#noinspection PyUnusedLocal
def start_response(status, headers=None):
return
USERDB = {
"diana":{
"birthdate": "02/14/2012",
"gender": "female"
}
}
BASE_ENVIRON = {'SERVER_PROTOCOL': 'HTTP/1.1',
'REQUEST_METHOD': 'GET',
'QUERY_STRING': '',
'HTTP_CONNECTION': 'keep-alive',
'REMOTE_ADDR': '127.0.0.1',
'wsgi.url_scheme': 'http',
'SERVER_PORT': '8087',
'PATH_INFO': '/register',
'HTTP_HOST': 'localhost:8087',
'HTTP_ACCEPT': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'HTTP_ACCEPT_LANGUAGE': 'sv-se',
'CONTENT_TYPE': 'text/plain',
'REMOTE_HOST': '1.0.0.127.in-addr.arpa',
'HTTP_ACCEPT_ENCODING': 'gzip, deflate',
'COMMAND_MODE': 'unix2003'}
CDB = {
"client_1": { "client_secret": "hemlig"}
}
def verify_client(env, req, cdb):
return True
FUNCTIONS = {
"verify_client": verify_client,
"userinfo": user_info
}
cc = ClaimsClient(client_id="client_1")
cc.client_secret="hemlig"
req = cc.construct_UserClaimsRequest(request_args={"user_id": "diana",
"claims_names":["gender", "birthdate"]})
srv = ClaimsServer("name", None, CDB, FUNCTIONS, USERDB)
srv.keyjar[""] = [KeyChain(source="file://rsa.key", usage=["ver", "sig"])]
assert srv
environ = BASE_ENVIRON.copy()
environ["REQUEST_METHOD"] = "POST"
txt = req.to_urlencoded()
environ["CONTENT_LENGTH"] = len(txt)
fil = StringIO.StringIO(buf=txt)
environ["wsgi.input"] = fil
resp = srv.claims_endpoint(environ, start_response, LOG())
print resp
assert len(resp) == 1
ucr = UserClaimsResponse().deserialize(resp[0], "json")
ucr.verify(keyjar = srv.keyjar)
print ucr

22
tests/debug_keybundle.py Normal file
View File

@ -0,0 +1,22 @@
from oic.utils.keyio import KeyBundle, key_eq
__author__ = 'rolandh'
jwk_url = ["https://connect.openid4.us/connect4us.jwk", # edmund
"https://connect-op.heroku.com/jwk.json"] # nov
x509_url = ["https://connect-op.heroku.com/cert.pem"]
kc0 = KeyBundle(source=jwk_url[1], src_type="jwk", type="rsa", usage=["sig", "enc"])
kc1 = KeyBundle(source=x509_url[0], src_type="x509", type="rsa", usage=["sig", "enc"])
kc0.update()
print kc0
kc1.update()
print kc1
print key_eq(kc0.get("rsa")[0], kc1.get("rsa")[0])

37
tests/debug_keyio.py Normal file
View File

@ -0,0 +1,37 @@
from oic.oic import ProviderConfigurationResponse
from oic.utils.keyio import KeyJar
from requests import request
__author__ = 'rohe0002'
def http_request(url, **kwargs):
return request("GET", url, **kwargs)
info = {"registration_endpoint": "https://openidconnect.info/connect/register",
"userinfo_endpoint": "https://openidconnect.info/connect/userinfo",
"token_endpoint_auth_types_supported": ["client_secret_basic", "client_secret_post", "client_secret_jwt", "private_key_jwt"],
"jwk_url": "https://openidconnect.info/jwk/jwk.json",
"userinfo_algs_supported": ["HS256", "HS384", "HS512", "RS256", "RS384", "RS512"],
"user_id_types_supported": ["pairwise", "public"],
"scopes_supported": ["openid", "profile", "email", "address", "phone"],
"token_endpoint": "https://openidconnect.info/connect/token",
"id_token_algs_supported": ["HS256", "HS384", "HS512", "RS256", "RS384", "RS512"],
"version": "3.0",
"token_endpoint_auth_algs_supported": ["RS256", "RS384", "RS512"],
"request_object_algs_supported": ["HS256", "HS384", "HS512", "RS256", "RS384", "RS512"],
"response_types_supported": ["code", "token", "id_token", "id_token token", "code token", "code id_token", "code id_token token"],
"authorization_endpoint": "https://openidconnect.info/connect/authorize",
"acrs_supported": ["1"],
"check_id_endpoint": "https://openidconnect.info/connect/check_session",
"x509_url": "https://openidconnect.info/x509/cert.pem",
"issuer": "https://openidconnect.info"}
pi = ProviderConfigurationResponse(**info)
kj = KeyJar()
kj.load_keys(pi, pi["issuer"])
keys = kj.get("ver", "rsa", "https://openidconnect.info")
print keys

25
tests/debug_keyio_local.py Normal file
View File

@ -0,0 +1,25 @@
from jwkest.jws import alg2keytype
from oic.oic.message import IdToken
from oic.utils.keyio import KeyBundle, KeyJar
__author__ = 'rohe0002'
b0 = KeyBundle(source="http://localhost:8090/exports/jwk.json", src_type="jwk",
usage=["ver", "dec", "sig"])
b1 = KeyBundle(source="http://localhost:8090/exports/cert.pem", src_type="x509",
usage=["ver", "dec", "sig"])
print b0
print b1
kj = KeyJar()
kj["foobar"] = [b0, b1]
idt = IdToken().from_dict({"user_id": "diana", "aud": "uo5nowsdL3ck",
"iss": "https://localhost:8092", "acr": "2",
"exp": 1354442188, "iat": 1354359388})
ckey = kj.get_signing_key(alg2keytype("RS256"), "foobar")
_signed_jwt = idt.to_jwt(key=ckey, algorithm="RS256")

26
tests/debug_registration_endpoint.py Normal file
View File

@ -0,0 +1,26 @@
from oic.oic.message import RegistrationResponseCARS
from oic.oic.message import RegistrationRequest
from pinit import provider_init
from pinit import BASE_ENVIRON
from pinit import start_response
__author__ = 'rohe0002'
server = provider_init
req = RegistrationRequest(type="client_associate")
req["application_type"] = "web"
req["application_name"] = "My super service"
req["redirect_uris"] = ["http://example.com/authz"]
req["contact"] = ["foo@example.com"]
environ = BASE_ENVIRON.copy()
environ["QUERY_STRING"] = req.to_urlencoded()
resp = server.registration_endpoint(environ, start_response)
print resp
regresp = RegistrationResponseCARS().deserialize(resp[0], "json")
print regresp.keys()
#assert _eq(regresp.keys(), ['client_secret', 'expires_at', 'client_id'])

6
tests/debug_signed_id_token.py Normal file
View File

@ -0,0 +1,6 @@
from oic.oic import AuthorizationRequest
__author__ = 'rolandh'
areq = AuthorizationRequest(response_type="id_token",
scope="openid")

28
tests/dyn_test_keyio.py Normal file
View File

@ -0,0 +1,28 @@
from oic.oic.message import ProviderConfigurationResponse
from oic.utils.keyio import KeyJar
__author__ = 'rolandh'
PROVIDER_INFO = {
"registration_endpoint": "https://connect-op.heroku.com/connect/client",
"userinfo_endpoint": "https://connect-op.heroku.com/user_info",
"token_endpoint_auth_types_supported": "client_secret_basic",
"request_object_signing_alg_values_supported": "RS256",
"user_id_types_supported": ["public", "pairwise"],
"scopes_supported": ["openid", "profile", "email", "address", "phone"],
"token_endpoint": "https://connect-op.heroku.com/access_tokens",
"id_token_algs_supported": ["RS256"], "version": "3.0",
"jwk_url": "https://connect-op.heroku.com/jwk.json",
"response_types_supported": ["code", "token", "id_token", "code token",
"code id_token", "id_token token",
"code id_token token"],
"authorization_endpoint": "https://connect-op.heroku.com/authorizations/new",
"x509_url": "https://connect-op.heroku.com/cert.pem",
"issuer": "https://connect-op.heroku.com"}
def test_provider():
ks = KeyJar()
pcr = ProviderConfigurationResponse().from_dict(PROVIDER_INFO)
ks.load_keys(pcr, "https://connect-op.heroku.com")
assert ks["https://connect-op.heroku.com"]

4
tests/fakeoicsrv.py
View File

@ -94,9 +94,9 @@ class MyFakeOICServer(Server):
def authorization_endpoint(self, query):
req = self.parse_authorization_request(query=query)
sid = self.sdb.create_authz_session(user_id="user", areq=req)
sid = self.sdb.create_authz_session(sub="user", areq=req)
_info = self.sdb[sid]
_info["user_id"] = _info["local_user_id"]
_info["sub"] = _info["local_sub"]
if "code" in req["response_type"]:
if "token" in req["response_type"]:

10
tests/test_oic.py
View File

@ -64,7 +64,7 @@ KEYJ = KeyJar()
KEYJ[""] = [KC_RSA, KC_HMAC_S]
KEYJ["client_1"] = [KC_HMAC_VS]
IDTOKEN = IdToken(iss="http://oic.example.org/", user_id="user_id",
IDTOKEN = IdToken(iss="http://oic.example.org/", sub="user_id",
aud=CLIENT_ID, exp=utc_time_sans_frac()+86400,
nonce="N0nce",
iat=time.time())
@ -451,7 +451,7 @@ class TestOICClient():
resp = self.client.do_check_session_request(request_args=args)
assert resp.type() == "IdToken"
assert _eq(resp.keys(), ['nonce', 'user_id', 'aud', 'iss', 'exp', 'iat'])
assert _eq(resp.keys(), ['nonce', 'sub', 'aud', 'iss', 'exp', 'iat'])
def test_do_end_session_request(self):
self.client.redirect_uris = ["https://www.example.com/authz"]
@ -517,7 +517,7 @@ class TestOICClient():
def test_openid_request_with_request_2(self):
areq = self.client.construct_AuthorizationRequest(
request_args={"scope":"openid", "response_type":["code"]},
idtoken_claims={"claims": {"user_id": {"value":"248289761001"}}},
idtoken_claims={"claims": {"sub": {"value":"248289761001"}}},
)
print areq
@ -1051,7 +1051,7 @@ def test_parse_check_session_request():
srv.keyjar = KEYJ
request = srv.parse_check_session_request(query=CSREQ.to_urlencoded())
assert request.type() == "IdToken"
assert _eq(request.keys(),['nonce', 'user_id', 'aud', 'iss', 'exp', 'iat'])
assert _eq(request.keys(),['nonce', 'sub', 'aud', 'iss', 'exp', 'iat'])
assert request["aud"] == "client_1"
def test_parse_end_session_request():
@ -1088,7 +1088,7 @@ def test_make_id_token():
srv.keyjar = KEYJ
srv.keyjar["http://oic.example/rp"] = KC_RSA
session = {"user_id": "user0",
session = {"sub": "user0",
"client_id": "http://oic.example/rp"}
issuer= "http://oic.example/idp"
code = "abcdefghijklmnop"

24
tests/test_oic_provider.py
View File

@ -206,7 +206,7 @@ USERDB = {
"nickname": "Hasse",
"email": "hans@example.org",
"verified": False,
"user_id": "user"
"sub": "user"
}
}
@ -284,7 +284,7 @@ def test_server_authorization_endpoint_request():
"prompt": ["none"]}
req = AuthorizationRequest(**bib)
ic = {"claims": {"user_id": { "value":"username" }}}
ic = {"claims": {"sub": { "value":"username" }}}
_keys = server.keyjar.get_signing_key(type="rsa")
req["request"] = make_openid_request(req, _keys, idtoken_claims=ic,
algorithm="RS256")
@ -319,7 +319,7 @@ def test_server_authorization_endpoint_id_token():
sid = sdb.create_authz_session("username", AREQ)
_info = sdb[sid]
_user_info = IdToken(iss="https://foo.example.om", user_id="foo",
_user_info = IdToken(iss="https://foo.example.om", sub="foo",
aud=bib["client_id"], exp=epoch_in_a_while(minutes=10),
acr="2", nonce=bib["nonce"])
@ -494,7 +494,7 @@ def test_server_authenticated_2():
assert isinstance(part[2], IdToken)
assert (part[2].keys(),['acr', 'aud', 'c_hash', 'email', 'exp', 'iss',
'name', 'nickname', 'user_id'])
'name', 'nickname', 'sub'])
def test_server_authenticated_token():
server = provider_init
@ -571,7 +571,7 @@ def test_token_endpoint():
access_grant = _sdb.token(sid=sid)
_sdb[sid] = {
"oauth_state": "authz",
"user_id": "user_id",
"sub": "user_id",
"authzreq": "",
"client_id": CLIENT_ID,
"code": access_grant,
@ -613,7 +613,7 @@ def test_token_endpoint_unauth():
access_grant = _sdb.token(sid=sid)
_sdb[sid] = {
"oauth_state": "authz",
"user_id": "user_id",
"sub": "user_id",
"authzreq": "",
"client_id": "client_1",
"code": access_grant,
@ -716,8 +716,8 @@ def test_userinfo_endpoint():
resp3 = server.userinfo_endpoint(environ, start_response)
ident = OpenIDSchema().deserialize(resp3[0], "json")
print ident.keys()
assert _eq(ident.keys(), ['nickname', 'user_id', 'name', 'email'])
assert ident["user_id"] == USERDB["user"]["user_id"]
assert _eq(ident.keys(), ['nickname', 'sub', 'name', 'email'])
assert ident["sub"] == USERDB["user"]["sub"]
def test_check_session_endpoint():
server = provider_init
@ -726,7 +726,7 @@ def test_check_session_endpoint():
# server.keyjar["number5"] = KeyChain({"hmac":CDB["number5"]["client_secret"]},
# usage=["ver"])
session = {"user_id": "UserID", "client_id": "number5"}
session = {"sub": "UserID", "client_id": "number5"}
idtoken = server.id_token_as_signed_jwt(session)
csr = CheckSessionRequest(id_token=idtoken)
environ = BASE_ENVIRON.copy()
@ -736,7 +736,7 @@ def test_check_session_endpoint():
print info
idt = IdToken().deserialize(info[0], "json")
print idt.keys()
assert _eq(idt.keys(), ['user_id', 'aud', 'iss', 'acr', 'exp', 'iat'])
assert _eq(idt.keys(), ['sub', 'aud', 'iss', 'acr', 'exp', 'iat'])
assert idt["iss"] == server.name
def test_registration_endpoint():
@ -842,7 +842,7 @@ def test_registered_redirect_uri_without_query_component():
resp = provider._verify_redirect_uri(areq)
print resp
assert resp == None
assert resp is None
def test_registered_redirect_uri_with_query_component():
provider2 = Provider("FOOP", {}, {}, None, None)
@ -862,7 +862,7 @@ def test_registered_redirect_uri_with_query_component():
faulty = [
"http://example.org/cb",
"http://example.org/cb/foo",
"http://example.org/cb?got=you"
"http://example.org/cb?got=you",
"http://example.org/cb?foo=you"
]
correct = [

12
tests/test_sdb.py
View File

@ -144,15 +144,15 @@ def test_create_authz_session_with_sector_id():
print info_1
assert "id_token" not in info_1
assert "oidreq" in info_1
assert info_1["user_id"] != "user_id"
user_id1 = info_1["user_id"]
assert info_1["sub"] != "user_id"
user_id1 = info_1["sub"]
sdb.do_userid(sid5, uid, "http://example.net/si.jwt", "pairwise")
info_2 = sdb[sid5]
print info_2
assert info_2["user_id"] != "user_id"
assert info_2["user_id"] != user_id1
assert info_2["sub"] != "user_id"
assert info_2["sub"] != user_id1
def test_update_to_token():
sdb = SessionDB()
@ -163,7 +163,7 @@ def test_update_to_token():
print _dict.keys()
assert _eq(_dict.keys(), ['code', 'oauth_state', 'issued', 'expires_at',
'token_type', 'client_id', 'authzreq',
'refresh_token', 'local_user_id', 'user_id',
'refresh_token', 'local_sub', 'sub',
'access_token', 'expires_in', 'state',
'redirect_uri', 'code_used', 'scope',
'access_token_scope', 'revoked'])
@ -180,7 +180,7 @@ def test_update_to_token():
print _dict.keys()
assert _eq(_dict.keys(), ['code', 'oauth_state', 'issued', 'expires_at',
'token_type', 'client_id', 'authzreq',
'refresh_token', 'local_user_id', 'user_id',
'refresh_token', 'local_sub', 'sub',
'oidreq', 'access_token', 'expires_in', 'state',
'redirect_uri', 'code_used', 'id_token',
'scope', 'access_token_scope', 'revoked'])

10
tests/test_x_claims.py
View File

@ -87,14 +87,14 @@ def test_1():
cc = ClaimsClient(client_id="client_1")
cc.client_secret="hemlig"
req = cc.construct_UserClaimsRequest(request_args={"user_id": "norah",
req = cc.construct_UserClaimsRequest(request_args={"sub": "norah",
"claims_names":["gender", "birthdate"]})
print req
assert req.type() == "UserClaimsRequest"
assert _eq(req.keys(),['client_secret', 'claims_names', 'user_id',
assert _eq(req.keys(),['client_secret', 'claims_names', 'sub',
'client_id'])
assert req["user_id"] == "norah"
assert req["sub"] == "norah"
assert req["client_id"] == "client_1"
def test_c2():
@ -103,7 +103,7 @@ def test_c2():
cc.userclaims_endpoint = "https://example.com/claims"
request=UserClaimsRequest
method = "POST"
request_args = {"user_id": "norah", "claims_names":["gender", "birthdate"]}
request_args = {"sub": "norah", "claims_names":["gender", "birthdate"]}
cc.request_info(request, method=method, request_args=request_args)
@ -126,7 +126,7 @@ def test_srv2():
cc = ClaimsClient(client_id="client_1")
cc.client_secret="hemlig"
req = cc.construct_UserClaimsRequest(request_args={"user_id": "diana",
req = cc.construct_UserClaimsRequest(request_args={"sub": "diana",
"claims_names":["gender", "birthdate"]})
srv = ClaimsServer("name", None, CDB, FUNCTIONS, USERDB)