ungleich-public/uotp
6
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Fixed misconseption: Same name but different realm represent different people

Browse source 
Fixed tests/*
Added scripts/create-auth.py
This commit is contained in:
ahmadbilalkhalid 2019-11-11 22:48:20 +05:00
parent ea0e0aeeb3
commit 83756a4bbc
7 changed files with 209 additions and 219 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Pipfile
View file

@ -5,6 +5,7 @@ verify_ssl = true
[dev-packages] [dev-packages]
pep8 = "*" pep8 = "*"
pycodestyle = "*"
[packages] [packages]
flask = "*" flask = "*"

122
Pipfile.lock generated
View file

@ -1,7 +1,7 @@
{ {
"_meta": { "_meta": {
"hash": { "hash": {
"sha256": "881d2dd7e14f980c6d76875c7ecc1949989d040129622e51cfe6849217ece4c0" "sha256": "d2909a9d9dc49af377037be55265460f701aa85e2d601cc90d292275ea01fba6"
}, },
"pipfile-spec": 6, "pipfile-spec": 6,
"requires": { "requires": {
@ -32,10 +32,10 @@
}, },
"attrs": { "attrs": {
"hashes": [ "hashes": [
"sha256:ec20e7a4825331c1b5ebf261d111e16fa9612c1f7a5e1f884f12bd53a664dfd2", "sha256:08a96c641c3a74e44eb59afb61a24f2cb9f4d7188748e76ba4bb5edfa3cb7d1c",
"sha256:f913492e1663d3c36f502e5e9ba6cd13cf19d7fab50aa13239e420fef95e1396" "sha256:f7b7ce16570fe9965acd6d30101a28f62fb4a7f9e926b3bbc9b61f8b04247e72"
], ],
"version": "==19.2.0" "version": "==19.3.0"
}, },
"certifi": { "certifi": {
"hashes": [ "hashes": [
@ -87,40 +87,56 @@
}, },
"grpcio": { "grpcio": {
"hashes": [ "hashes": [
"sha256:0302331e014fc4bac028b6ad480b33f7abfe20b9bdcca7be417124dda8f22115", "sha256:0419ae5a45f49c7c40d9ae77ae4de9442431b7822851dfbbe56ee0eacb5e5654",
"sha256:0aa0cce9c5eb1261b32173a20ed42b51308d55ce28ecc2021e868b3cb90d9503", "sha256:1e8631eeee0fb0b4230aeb135e4890035f6ef9159c2a3555fa184468e325691a",
"sha256:0c83947575300499adbc308e986d754e7f629be0bdd9bea1ffdd5cf76e1f1eff", "sha256:24db2fa5438f3815a4edb7a189035051760ca6aa2b0b70a6a948b28bfc63c76b",
"sha256:0ca26ff968d45efd4ef73447c4d4b34322ea8c7d06fbb6907ce9e5db78f1bbcb", "sha256:2adb1cdb7d33e91069517b41249622710a94a1faece1fed31cd36904e4201cde",
"sha256:0cf80a7955760c2498f8821880242bb657d70998065ff0d2a082de5ffce230a7", "sha256:2cd51f35692b551aeb1fdeb7a256c7c558f6d78fcddff00640942d42f7aeba5f",
"sha256:0d40706e57d9833fe0e023a08b468f33940e8909affa12547874216d36bba208", "sha256:3247834d24964589f8c2b121b40cd61319b3c2e8d744a6a82008643ef8a378b1",
"sha256:11872069156de34c6f3f9a1deb46cc88bc35dfde88262c4c73eb22b39b16fc55", "sha256:3433cb848b4209717722b62392e575a77a52a34d67c6730138102abc0a441685",
"sha256:16065227faae0ab0abf1789bfb92a2cd2ab5da87630663f93f8178026da40e0d", "sha256:39671b7ff77a962bd745746d9d2292c8ed227c5748f16598d16d8631d17dd7e5",
"sha256:1e33778277685f6fabb22539136269c87c029e39b6321ef1a639b756a1c0a408", "sha256:40a0b8b2e6f6dd630f8b267eede2f40a848963d0f3c40b1b1f453a4a870f679e",
"sha256:2b16be15b1ae656bc7a36642b8c7045be2dde2048bb4b67478003e9d9db8022a", "sha256:40f9a74c7aa210b3e76eb1c9d56aa8d08722b73426a77626967019df9bbac287",
"sha256:3701dfca3ada27ceef0d17f728ce9dfef155ed20c57979c2b05083082258c6c1", "sha256:423f76aa504c84cb94594fb88b8a24027c887f1c488cf58f2173f22f4fbd046c",
"sha256:41912ecaf482abf2de74c69f509878f99223f5dd6b2de1a09c955afd4de3cf9b", "sha256:43bd04cec72281a96eb361e1b0232f0f542b46da50bcfe72ef7e5a1b41d00cb3",
"sha256:4332cbd20544fe7406910137590f38b5b3a1f6170258e038652cf478c639430f", "sha256:43e38762635c09e24885d15e3a8e374b72d105d4178ee2cc9491855a8da9c380",
"sha256:44068ecbdc6467c2bff4d8198816c8a2701b6dd1ec16078fceb6adc7c1f577d6", "sha256:4413b11c2385180d7de03add6c8845dd66692b148d36e27ec8c9ef537b2553a1",
"sha256:53115960e37059420e2d16a4b04b00dd2ab3b6c3c67babd01ffbfdcd7881a69b", "sha256:4450352a87094fd58daf468b04c65a9fa19ad11a0ac8ac7b7ff17d46f873cbc1",
"sha256:6e7027bcd4070414751e2a5e60706facb98a1fc636497c9bac5442fe37b8ae6b", "sha256:49ffda04a6e44de028b3b786278ac9a70043e7905c3eea29eed88b6524d53a29",
"sha256:6ff57fb2f07b7226b5bec89e8e921ea9bd220f35f11e094f2ba38f09eecd49c6", "sha256:4a38c4dde4c9120deef43aaabaa44f19186c98659ce554c29788c4071ab2f0a4",
"sha256:73240e244d7644654bbda1f309f4911748b6a1804b7a8897ddbe8a04c90f7407", "sha256:50b1febdfd21e2144b56a9aa226829e93a79c354ef22a4e5b013d9965e1ec0ed",
"sha256:785234bbc469bc75e26c868789a2080ffb30bd6e93930167797729889ad06b0b", "sha256:559b1a3a8be7395ded2943ea6c2135d096f8cc7039d6d12127110b6496f251fe",
"sha256:82f9d3c7f91d2d1885631335c003c5d45ae1cd69cc0bc4893f21fef50b8151bc", "sha256:5de86c182667ec68cf84019aa0d8ceccf01d352cdca19bf9e373725204bdbf50",
"sha256:86bdc2a965510658407a1372eb61f0c92f763fdfb2795e4d038944da4320c950", "sha256:5fc069bb481fe3fad0ba24d3baaf69e22dfa6cc1b63290e6dfeaf4ac1e996fb7",
"sha256:95e925b56676a55e6282b3de80a1cbad5774072159779c61eac02791dface049", "sha256:6a19d654da49516296515d6f65de4bbcbd734bc57913b21a610cfc45e6df3ff1",
"sha256:96673bb4f14bd3263613526d1e7e33fdb38a9130e3ce87bf52314965706e1900", "sha256:7535b3e52f498270e7877dde1c8944d6b7720e93e2e66b89c82a11447b5818f5",
"sha256:970014205e76920484679035b6fb4b16e02fc977e5aac4d22025da849c79dab9", "sha256:7c4e495bcabc308198b8962e60ca12f53b27eb8f03a21ac1d2d711d6dd9ecfca",
"sha256:ace5e8bf11a1571f855f5dab38a9bd34109b6c9bc2864abf24a597598c7e3695", "sha256:8a8fc4a0220367cb8370cedac02272d574079ccc32bffbb34d53aaf9e38b5060",
"sha256:ad375f03eb3b9cb75a24d91eab8609e134d34605f199efc41e20dd642bdac855", "sha256:8b008515e067232838daca020d1af628bf6520c8cc338bf383284efe6d8bd083",
"sha256:b819c4c7dcf0de76788ce5f95daad6d4e753d6da2b6a5f84e5bb5b5ce95fddc4", "sha256:8d1684258e1385e459418f3429e107eec5fb3d75e1f5a8c52e5946b3f329d6ea",
"sha256:c17943fd340cbd906db49f3f03c7545e5a66b617e8348b2c7a0d2c759d216af1", "sha256:8eb5d54b87fb561dc2e00a5c5226c33ffe8dbc13f2e4033a412bafb7b37b194d",
"sha256:d21247150dea86dabd3b628d8bc4b563036db3d332b3f4db3c5b1b0b122cb4f6", "sha256:94cdef0c61bd014bb7af495e21a1c3a369dd0399c3cd1965b1502043f5c88d94",
"sha256:d4d500a7221116de9767229ff5dd10db91f789448d85befb0adf5a37b0cd83b5", "sha256:9d9f3be69c7a5e84c3549a8c4403fa9ac7672da456863d21e390b2bbf45ccad1",
"sha256:e2a942a3cfccbbca21a90c144867112698ef36486345c285da9e98c466f22b22", "sha256:9fb6fb5975a448169756da2d124a1beb38c0924ff6c0306d883b6848a9980f38",
"sha256:e983273dca91cb8a5043bc88322eb48e2b8d4e4998ff441a1ee79ced89db3909" "sha256:a5eaae8700b87144d7dfb475aa4675e500ff707292caba3deff41609ddc5b845",
"sha256:aaeac2d552772b76d24eaff67a5d2325bc5205c74c0d4f9fbe71685d4a971db2",
"sha256:bb611e447559b3b5665e12a7da5160c0de6876097f62bf1d23ba66911564868e",
"sha256:bc0d41f4eb07da8b8d3ea85e50b62f6491ab313834db86ae2345be07536a4e5a",
"sha256:bf51051c129b847d1bb63a9b0826346b5f52fb821b15fe5e0d5ef86f268510f5",
"sha256:c948c034d8997526011960db54f512756fb0b4be1b81140a15b4ef094c6594a4",
"sha256:d435a01334157c3b126b4ee5141401d44bdc8440993b18b05e2f267a6647f92d",
"sha256:d46c1f95672b73288e08cdca181e14e84c6229b5879561b7b8cfd48374e09287",
"sha256:d5d58309b42064228b16b0311ff715d6c6e20230e81b35e8d0c8cfa1bbdecad8",
"sha256:dc6e2e91365a1dd6314d615d80291159c7981928b88a4c65654e3fefac83a836",
"sha256:e0dfb5f7a39029a6cbec23affa923b22a2c02207960fd66f109e01d6f632c1eb",
"sha256:eb4bf58d381b1373bd21d50837a53953d625d1693f1b58fed12743c75d3dd321",
"sha256:ebb211a85248dbc396b29320273c1ffde484b898852432613e8df0164c091006",
"sha256:ec759ece4786ae993a5b7dc3b3dead6e9375d89a6c65dfd6860076d2eb2abe7b",
"sha256:f55108397a8fa164268238c3e69cc134e945d1f693572a2f05a028b8d0d2b837",
"sha256:f6c706866d424ff285b85a02de7bbe5ed0ace227766b2c42cbe12f3d9ea5a8aa",
"sha256:f8370ad332b36fbad117440faf0dd4b910e80b9c49db5648afd337abdde9a1b6"
], ],
"version": "==1.24.1" "version": "==1.25.0"
}, },
"idna": { "idna": {
"hashes": [ "hashes": [
@ -243,18 +259,18 @@
}, },
"pyparsing": { "pyparsing": {
"hashes": [ "hashes": [
"sha256:6f98a7b9397e206d78cc01df10131398f1c8b8510a2f4d97d9abd82e1aacdd80", "sha256:4acadc9a2b96c19fe00932a38ca63e601180c39a189a696abce1eaab641447e1",
"sha256:d9338df12903bbf5d65a0e4e87c2161968b10d2e489652bb47001d82a9b028b4" "sha256:61b5ed888beab19ddccab3478910e2076a6b5a0295dffc43021890e136edf764"
], ],
"version": "==2.4.2" "version": "==2.4.4"
}, },
"pytest": { "pytest": {
"hashes": [ "hashes": [
"sha256:7e4800063ccfc306a53c461442526c5571e1462f61583506ce97e4da6a1d88c8", "sha256:27abc3fef618a01bebb1f0d6d303d2816a99aa87a5968ebc32fe971be91eb1e6",
"sha256:ca563435f4941d0cb34767301c27bc65c510cb82e90b9ecf9cb52dc2c63caaa0" "sha256:58cee9e09242937e136dbb3dab466116ba20d6b7828c7620f23947f37eb4dae4"
], ],
"index": "pypi", "index": "pypi",
"version": "==5.2.1" "version": "==5.2.2"
}, },
"python-decouple": { "python-decouple": {
"hashes": [ "hashes": [
@ -280,17 +296,17 @@
}, },
"six": { "six": {
"hashes": [ "hashes": [
"sha256:3350809f0555b11f552448330d0b52d5f24c91a322ea4a15ef22629740f3761c", "sha256:1f1b7d42e254082a9db6279deae68afb421ceba6158efa6131de7b3003ee93fd",
"sha256:d16a0141ec1a18405cd4ce8b4613101da75da0e9a7aec5bdd4fa804d0e0eba73" "sha256:30f610279e8b2578cab6db20741130331735c781b56053c59c4076da27f06b66"
], ],
"version": "==1.12.0" "version": "==1.13.0"
}, },
"tenacity": { "tenacity": {
"hashes": [ "hashes": [
"sha256:6a7511a59145c2e319b7d04ddd93c12d48cc3d3c8fa42c2846d33a620ee91f57", "sha256:72f397c2bb1887e048726603f3f629ea16f88cb3e61e4ed3c57e98582b8e3571",
"sha256:a4eb168dbf55ed2cae27e7c6b2bd48ab54dabaf294177d998330cf59f294c112" "sha256:947e728aedf06e8db665bb7898112e90d17e48cc3f3289784a2b9ccf6e56fabc"
], ],
"version": "==5.1.1" "version": "==6.0.0"
}, },
"urllib3": { "urllib3": {
"hashes": [ "hashes": [
@ -329,6 +345,14 @@
], ],
"index": "pypi", "index": "pypi",
"version": "==1.7.1" "version": "==1.7.1"
},
"pycodestyle": {
"hashes": [
"sha256:95a2219d12372f05704562a14ec30bc76b05a5b297b21a5dfe3f6fac3491ae56",
"sha256:e40a936c9a450ad81df37f549d676d127b1b66000a6c500caa2b085bc0ca976c"
],
"index": "pypi",
"version": "==2.5.0"
} }
} }
} }

38
app.py
View file

@ -6,12 +6,7 @@ from os.path import join
from flask import Flask, request from flask import Flask, request
from flask_restful import Resource, Api from flask_restful import Resource, Api
from schemas import ( from schemas import OTPSchema, CreateOTPSchema, DeleteOTPSchema, ListAccountSchema
OTPSchema,
CreateOTPSchema,
DeleteOTPSchema,
ListAccountSchema,
)
from config import etcd_client from config import etcd_client
from helper import is_valid_otp, create_admin_if_dont_exists from helper import is_valid_otp, create_admin_if_dont_exists
@ -25,28 +20,30 @@ class Verify(Resource):
@staticmethod @staticmethod
def get(): def get():
data = request.json data = request.json
if data:
schema = OTPSchema(data) schema = OTPSchema(data)
if schema.is_valid(): if schema.is_valid():
return {"message": "Verified"}, 200 return {"message": "Verified"}, 200
else: else:
return schema.get_errors(), 400 return schema.get_errors(), 400
else:
return {"message": "No Data"}, 400
class Create(Resource): class Create(Resource):
@staticmethod @staticmethod
def post(): def post():
data = request.json data = request.json
schema = CreateOTPSchema(data) schema = CreateOTPSchema(data)
if schema.is_valid(): if schema.is_valid():
_key = join( _key = join(decouple.config("BASE_PREFIX"), data["name"])
decouple.config("BASE_PREFIX"),
data["realm"],
data["name"],
)
if etcd_client.get(_key) is None: if etcd_client.get(_key) is None:
_value = {"seed": pyotp.random_base32()} if not isinstance(data["realm"], list):
realms = [data["realm"]]
else:
realms = data["realm"]
_value = {"seed": pyotp.random_base32(), "realm": realms}
etcd_client.put(_key, _value, value_in_json=True) etcd_client.put(_key, _value, value_in_json=True)
return { return {
"message": "Account Created\n" "message": "Account Created\n"
@ -65,11 +62,7 @@ class Delete(Resource):
schema = DeleteOTPSchema(data) schema = DeleteOTPSchema(data)
if schema.is_valid(): if schema.is_valid():
_key = join( _key = join(decouple.config("BASE_PREFIX"), data["name"])
decouple.config("BASE_PREFIX"),
data["realm"],
data["name"],
)
etcd_client.client.delete(_key) etcd_client.client.delete(_key)
return {"message": "Account Deleted"} return {"message": "Account Deleted"}
@ -89,8 +82,11 @@ class List(Resource):
) )
r = {} r = {}
for entry in result: for entry in result:
_realm, _name = entry.key.split("/")[-2:] _name = entry.key.split("/")[-1]
r['{}/{}'.format(_realm, _name)] = entry.value['seed'] r["{}".format(_name)] = {
"seed": entry.value["seed"],
"realm": entry.value["realm"],
}
return r return r
else: else:
return schema.get_errors(), 400 return schema.get_errors(), 400
@ -102,4 +98,4 @@ api.add_resource(Delete, "/delete")
api.add_resource(List, "/list") api.add_resource(List, "/list")
if __name__ == "__main__": if __name__ == "__main__":
app.run(debug=True) app.run(debug=True, port=decouple.config("PORT", int))

17
helper.py
View file

@ -1,12 +1,16 @@
import pyotp import pyotp
import decouple import decouple
import requests
from os.path import join from os.path import join
def is_valid_otp(etcd_client, name, realm, token): def is_valid_otp(etcd_client, name, realm, token):
_key = join(decouple.config("BASE_PREFIX"), realm, name) _key = join(decouple.config("BASE_PREFIX"), name)
entry = etcd_client.get(_key, value_in_json=True) entry = etcd_client.get(_key, value_in_json=True)
if entry: if entry:
if realm not in entry.value["realm"]:
return False
totp = pyotp.TOTP(entry.value["seed"]) totp = pyotp.TOTP(entry.value["seed"])
try: try:
is_token_valid = totp.verify(token) is_token_valid = totp.verify(token)
@ -19,12 +23,11 @@ def is_valid_otp(etcd_client, name, realm, token):
def create_admin_if_dont_exists(etcd_client): def create_admin_if_dont_exists(etcd_client):
_key = join( _key = join(decouple.config("BASE_PREFIX"), "admin")
decouple.config("BASE_PREFIX"),
decouple.config("ADMIN_REALM"),
"admin",
)
if etcd_client.get(_key) is None: if etcd_client.get(_key) is None:
print("admin does not exists!. So, creating one") print("admin does not exists!. So, creating one")
_value = {"seed": pyotp.random_base32()} _value = {
"seed": pyotp.random_base32(),
"realm": [decouple.config("ADMIN_REALM")],
}
etcd_client.put(_key, _value, value_in_json=True) etcd_client.put(_key, _value, value_in_json=True)

137
schemas.py
View file

@ -6,6 +6,7 @@ from os.path import join
from helper import is_valid_otp from helper import is_valid_otp
from config import etcd_client from config import etcd_client
from typing import Union
class Field: class Field:
@ -20,16 +21,15 @@ class Field:
def is_valid(self): def is_valid(self):
if self.value == KeyError: if self.value == KeyError:
self.add_error( self.add_error("'{}' field is a required field".format(self.name))
"'{}' field is a required field".format(self.name)
)
else: else:
if not isinstance(self.value, self.type): try:
self.add_error( _type = self.type.__args__
"Incorrect Type for '{}' field".format( except Exception:
self.name _type = self.type
)
) if not isinstance(self.value, _type):
self.add_error("Incorrect Type for '{}' field".format(self.name))
else: else:
self.validation() self.validation()
@ -46,7 +46,6 @@ class Field:
class BaseSchema: class BaseSchema:
def __init__(self, data, fields=None): def __init__(self, data, fields=None):
_ = data # suppress linter warning
self.__errors = [] self.__errors = []
if fields is None: if fields is None:
self.fields = [] self.fields = []
@ -95,22 +94,12 @@ class DataRequiredSchema(BaseSchema):
class OTPSchema(DataRequiredSchema): class OTPSchema(DataRequiredSchema):
def __init__(self, data: dict, fields=None): def __init__(self, data: dict, fields=None):
self.name = Field("name", str, data.get("name", KeyError)) self.name = Field("name", str, data.get("name", KeyError))
self.realm = Field( self.realm = Field("realm", str, data.get("realm", KeyError))
"realm", str, data.get("realm", KeyError) self.token = Field("token", str, data.get("token", KeyError))
)
self.token = Field(
"token", str, data.get("token", KeyError)
)
self.auth_name = Field( self.auth_name = Field("auth_name", str, data.get("auth_name", KeyError))
"auth-name", str, data.get("auth-name", KeyError) self.auth_realm = Field("auth_realm", str, data.get("auth_realm", KeyError))
) self.auth_token = Field("auth_token", str, data.get("auth_token", KeyError))
self.auth_realm = Field(
"auth-realm", str, data.get("auth-realm", KeyError)
)
self.auth_token = Field(
"auth-token", str, data.get("auth-token", KeyError)
)
self.auth_realm.validation = self.auth_realm_validation self.auth_realm.validation = self.auth_realm_validation
_fields = [ _fields = [
self.name, self.name,
@ -128,9 +117,7 @@ class OTPSchema(DataRequiredSchema):
def auth_realm_validation(self): def auth_realm_validation(self):
if self.auth_realm.value != decouple.config("AUTH_REALM"): if self.auth_realm.value != decouple.config("AUTH_REALM"):
self.add_error( self.add_error(
"Authentication realm must be {}".format( "Authentication realm must be {}".format(decouple.config("AUTH_REALM"))
decouple.config("AUTH_REALM")
)
) )
def validation(self): def validation(self):
@ -142,16 +129,9 @@ class OTPSchema(DataRequiredSchema):
): ):
if is_valid_otp( if is_valid_otp(
etcd_client, etcd_client, self.name.value, self.realm.value, self.token.value
self.name.value,
self.realm.value,
self.token.value,
): ):
_key = join( _key = join(decouple.config("BASE_PREFIX"), self.name.value)
decouple.config("BASE_PREFIX"),
self.realm.value,
self.name.value,
)
entry = etcd_client.get(_key, value_in_json=True) entry = etcd_client.get(_key, value_in_json=True)
if not entry: if not entry:
self.add_error("No such Account Found") self.add_error("No such Account Found")
@ -164,19 +144,11 @@ class OTPSchema(DataRequiredSchema):
class CreateOTPSchema(DataRequiredSchema): class CreateOTPSchema(DataRequiredSchema):
def __init__(self, data: dict, fields=None): def __init__(self, data: dict, fields=None):
self.name = Field("name", str, data.get("name", KeyError)) self.name = Field("name", str, data.get("name", KeyError))
self.realm = Field( self.realm = Field("realm", Union[str, list], data.get("realm", KeyError))
"realm", str, data.get("realm", KeyError)
)
self.admin_name = Field( self.admin_name = Field("admin_name", str, data.get("admin_name", KeyError))
"admin-name", str, data.get("admin-name", KeyError) self.admin_realm = Field("admin_realm", str, data.get("admin_realm", KeyError))
) self.admin_token = Field("admin_token", str, data.get("admin_token", KeyError))
self.admin_realm = Field(
"admin-realm", str, data.get("admin-realm", KeyError)
)
self.admin_token = Field(
"admin-token", str, data.get("admin-token", KeyError)
)
self.admin_realm.validation = self.admin_realm_validation self.admin_realm.validation = self.admin_realm_validation
_fields = [ _fields = [
@ -193,9 +165,7 @@ class CreateOTPSchema(DataRequiredSchema):
def admin_realm_validation(self): def admin_realm_validation(self):
if self.admin_realm.value != decouple.config("ADMIN_REALM"): if self.admin_realm.value != decouple.config("ADMIN_REALM"):
self.add_error( self.add_error(
"Admin must be from {} realm".format( "Admin must be from {} realm".format(decouple.config("ADMIN_REALM"))
decouple.config("ADMIN_REALM")
)
) )
def validation(self): def validation(self):
@ -206,11 +176,7 @@ class CreateOTPSchema(DataRequiredSchema):
self.admin_token.value, self.admin_token.value,
): ):
_key = join( _key = join(decouple.config("BASE_PREFIX"), self.name.value)
decouple.config("BASE_PREFIX"),
self.realm.value,
self.name.value,
)
if etcd_client.get(_key): if etcd_client.get(_key):
self.add_error("Account already exists") self.add_error("Account already exists")
else: else:
@ -220,28 +186,13 @@ class CreateOTPSchema(DataRequiredSchema):
class DeleteOTPSchema(DataRequiredSchema): class DeleteOTPSchema(DataRequiredSchema):
def __init__(self, data: dict, fields=None): def __init__(self, data: dict, fields=None):
self.name = Field("name", str, data.get("name", KeyError)) self.name = Field("name", str, data.get("name", KeyError))
self.realm = Field(
"realm", str, data.get("realm", KeyError)
)
self.admin_name = Field( self.admin_name = Field("admin_name", str, data.get("admin_name", KeyError))
"admin-name", str, data.get("admin-name", KeyError) self.admin_realm = Field("admin_realm", str, data.get("admin_realm", KeyError))
) self.admin_token = Field("admin_token", str, data.get("admin_token", KeyError))
self.admin_realm = Field(
"admin-realm", str, data.get("admin-realm", KeyError)
)
self.admin_token = Field(
"admin-token", str, data.get("admin-token", KeyError)
)
self.admin_realm.validation = self.admin_realm_validation self.admin_realm.validation = self.admin_realm_validation
_fields = [ _fields = [self.name, self.admin_name, self.admin_realm, self.admin_token]
self.name,
self.realm,
self.admin_name,
self.admin_realm,
self.admin_token,
]
if fields: if fields:
_fields += fields _fields += fields
super().__init__(data=data, fields=_fields) super().__init__(data=data, fields=_fields)
@ -249,9 +200,7 @@ class DeleteOTPSchema(DataRequiredSchema):
def admin_realm_validation(self): def admin_realm_validation(self):
if self.admin_realm.value != decouple.config("ADMIN_REALM"): if self.admin_realm.value != decouple.config("ADMIN_REALM"):
self.add_field_errors( self.add_field_errors(
"Admin must be from {} realm".format( "Admin must be from {} realm".format(decouple.config("ADMIN_REALM"))
decouple.config("ADMIN_REALM")
)
) )
def validation(self): def validation(self):
@ -262,11 +211,7 @@ class DeleteOTPSchema(DataRequiredSchema):
self.admin_token.value, self.admin_token.value,
): ):
_key = join( _key = join(decouple.config("BASE_PREFIX"), self.name.value)
decouple.config("BASE_PREFIX"),
self.realm.value,
self.name.value,
)
if not etcd_client.get(_key): if not etcd_client.get(_key):
self.add_error("Account does not exists") self.add_error("Account does not exists")
else: else:
@ -275,23 +220,13 @@ class DeleteOTPSchema(DataRequiredSchema):
class ListAccountSchema(DataRequiredSchema): class ListAccountSchema(DataRequiredSchema):
def __init__(self, data: dict, fields=None): def __init__(self, data: dict, fields=None):
data = data or {'': None} data = data or {"": None}
self.admin_name = Field( self.admin_name = Field("admin_name", str, data.get("admin_name", KeyError))
"admin-name", str, data.get("admin-name", KeyError) self.admin_realm = Field("admin_realm", str, data.get("admin_realm", KeyError))
) self.admin_token = Field("admin_token", str, data.get("admin_token", KeyError))
self.admin_realm = Field(
"admin-realm", str, data.get("admin-realm", KeyError)
)
self.admin_token = Field(
"admin-token", str, data.get("admin-token", KeyError)
)
self.admin_realm.validation = self.admin_realm_validation self.admin_realm.validation = self.admin_realm_validation
_fields = [ _fields = [self.admin_name, self.admin_realm, self.admin_token]
self.admin_name,
self.admin_realm,
self.admin_token,
]
if fields: if fields:
_fields += fields _fields += fields
super().__init__(data=data, fields=_fields) super().__init__(data=data, fields=_fields)
@ -299,9 +234,7 @@ class ListAccountSchema(DataRequiredSchema):
def admin_realm_validation(self): def admin_realm_validation(self):
if self.admin_realm.value != decouple.config("ADMIN_REALM"): if self.admin_realm.value != decouple.config("ADMIN_REALM"):
self.add_field_errors( self.add_field_errors(
"Admin must be from {} realm".format( "Admin must be from {} realm".format(decouple.config("ADMIN_REALM"))
decouple.config("ADMIN_REALM")
)
) )
def validation(self): def validation(self):

19
scripts/create-auth.py Normal file
View file

@ -0,0 +1,19 @@
import requests
import pyotp
import json
import decouple
import argparse
arg_parser = argparse.ArgumentParser()
arg_parser.add_argument("admin_seed", required=True)
args = arg_parser.parse_args()
r = requests.post("http://localhost:{}/create".format(decouple.config("PORT")),
json={
"name": "auth",
"realm": ["ungleich-auth"],
"admin_name": "admin",
"admin_realm": "ungleich-admin",
"admin_token": pyotp.TOTP(args.admin_seed).now()
})
print(json.loads(r.content.decode("utf-8")))

74
tests/test_uotp.py
View file

@ -35,7 +35,6 @@ class TestUOTP(unittest.TestCase):
entry = TestUOTP.etcd_client.get( entry = TestUOTP.etcd_client.get(
os.path.join( os.path.join(
os.environ['BASE_PREFIX'], os.environ['BASE_PREFIX'],
os.environ['ADMIN_REALM'],
'admin' 'admin'
), value_in_json=True ), value_in_json=True
) )
@ -56,46 +55,45 @@ class TestUOTP(unittest.TestCase):
return super().tearDownClass() return super().tearDownClass()
def get_otp_list(self): def get_otp_list(self):
r = self.app_client.get('/list', json_data = {
json={ 'admin_name': self.admin_name,
'admin-name': self.admin_name, 'admin_realm': self.admin_realm,
'admin-realm': self.admin_realm, 'admin_token': pyotp.TOTP(self.admin_seed).now()
'admin-token': pyotp.TOTP(self.admin_seed).now()
} }
) r = self.app_client.get('/list', json=json_data)
return r return r
def create_otp(self, _name, _realm, _admin_name, _admin_realm, _admin_seed): def create_otp(self, _name, _realm,
r = self.app_client.post('/create', _admin_name, _admin_realm, _admin_seed):
json={
json_data = {
"name": _name, "name": _name,
"realm": _realm, "realm": _realm,
"admin-name": _admin_name, "admin_name": _admin_name,
"admin-realm": _admin_realm, "admin_realm": _admin_realm,
"admin-token": pyotp.TOTP(_admin_seed).now() "admin_token": pyotp.TOTP(_admin_seed).now()
}) }
r = self.app_client.post('/create', json=json_data)
return r return r
def verify_otp(self, _name, _realm, _seed, def verify_otp(self, _name, _realm, _seed,
_auth_name, _auth_realm, _auth_seed): _auth_name, _auth_realm, _auth_seed):
r = self.app_client.get('/verify',
json={ json_data = {
"name": _name, "name": _name,
"realm": _realm, "realm": _realm,
"token": pyotp.TOTP(_seed).now(), "token": pyotp.TOTP(_seed).now(),
"auth-name": _auth_name, "auth_name": _auth_name,
"auth-realm": _auth_realm, "auth_realm": _auth_realm,
"auth-token": pyotp.TOTP(_auth_seed).now() "auth_token": pyotp.TOTP(_auth_seed).now()
}) }
r = self.app_client.get('/verify', json=json_data)
return r return r
def test_list(self): def test_list(self):
r = self.get_otp_list() r = self.get_otp_list()
self.assertEqual(r.status_code, 200) self.assertEqual(r.status_code, 200)
self.assertIn( self.assertIn(self.admin_name, r.json)
'{}/{}'.format(self.admin_realm, self.admin_name),
r.json
)
def test_create(self): def test_create(self):
_name = 'auth' _name = 'auth'
@ -109,10 +107,7 @@ class TestUOTP(unittest.TestCase):
self.assertEqual(r.status_code, 200) self.assertEqual(r.status_code, 200)
r = self.get_otp_list() r = self.get_otp_list()
self.assertIn( self.assertIn(_name, r.json)
'{}/{}'.format(_realm, _name),
r.json
)
# Test Unsuccesful Creation i.e User from non-admin realm # Test Unsuccesful Creation i.e User from non-admin realm
# tries to create OTP account # tries to create OTP account
@ -120,7 +115,7 @@ class TestUOTP(unittest.TestCase):
# Get Auth Account # Get Auth Account
entry = self.etcd_client.get( entry = self.etcd_client.get(
os.path.join( os.path.join(
os.environ['BASE_PREFIX'], _realm, _name os.environ['BASE_PREFIX'], _name
), value_in_json=True ), value_in_json=True
) )
_adversery_name = 'adversery' _adversery_name = 'adversery'
@ -149,7 +144,6 @@ class TestUOTP(unittest.TestCase):
entry = self.etcd_client.get( entry = self.etcd_client.get(
os.path.join( os.path.join(
os.environ['BASE_PREFIX'], os.environ['BASE_PREFIX'],
_auth_realm,
_auth_name _auth_name
), ),
value_in_json=True value_in_json=True
@ -177,7 +171,27 @@ class TestUOTP(unittest.TestCase):
self.admin_realm, self.admin_seed) self.admin_realm, self.admin_seed)
self.assertEqual(r.status_code, 400) self.assertEqual(r.status_code, 400)
def test_delete(self):
_name = "protem"
_realm = "protem-realm"
r = self.create_otp(_name, _realm,
self.admin_name,
self.admin_realm,
self.admin_seed)
self.assertEqual(r.status_code, 200)
json_data = {
"name": _name,
"realm": _realm,
'admin_name': self.admin_name,
'admin_realm': self.admin_realm,
'admin_token': pyotp.TOTP(self.admin_seed).now()
}
r = self.app_client.post('/delete', json=json_data)
self.assertEqual(r.status_code, 200)
self.assertNotIn(_name, self.get_otp_list().get_json())
if __name__ == '__main__': if __name__ == '__main__':