trustchain-oc1-orchestral
/
authentik
Archived
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #42 from BeryJu/e2e 

e2e tests
This commit is contained in:
Jens L 2020-06-25 18:35:59 +02:00 committed by GitHub
parent 2dfec43750 5a11206fe9
commit a2ed53c312
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
90 changed files with 2742 additions and 308 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
.github/workflows/ci.yml vendored
View File

@ -121,10 +121,28 @@ jobs:
- uses: actions/setup-python@v1
with:
python-version: '3.8'
- uses: actions/setup-node@v1
with:
node-version: '12'
- name: Install dependencies
run: sudo pip install -U wheel pipenv && pipenv install --dev
run: |
sudo pip install -U wheel pipenv
pipenv install --dev
- name: Prepare Chrome node
run: |
cd e2e
docker-compose pull -q chrome
docker-compose up -d chrome
- name: Build static files for e2e test
run: |
cd passbook/static/static
yarn
- name: Run coverage
run: pipenv run ./scripts/coverage.sh
run: pipenv run coverage run ./manage.py test --failfast
- uses: actions/upload-artifact@v2
if: failure()
with:
path: out/
- name: Create XML Report
run: pipenv run coverage xml
- uses: codecov/codecov-action@v1

3
.github/workflows/release.yml vendored
View File

@ -82,8 +82,7 @@ jobs:
- uses: actions/checkout@v1
- name: Run test suite in final docker images
run: |
export PASSBOOK_DOMAIN=localhost
docker-compose pull
docker-compose pull -q
docker-compose up --no-start
docker-compose start postgresql redis
docker-compose run -u root server bash -c "pip install --no-cache -r requirements-dev.txt && ./manage.py test"

3
.github/workflows/tag.yml vendored
View File

@ -13,8 +13,7 @@ jobs:
- uses: actions/checkout@master
- name: Pre-release test
run: |
export PASSBOOK_DOMAIN=localhost
docker-compose pull
docker-compose pull -q
docker build \
--no-cache \
-t beryju/passbook:latest \

3
Pipfile
View File

@ -40,6 +40,7 @@ signxml = "*"
structlog = "*"
swagger-spec-validator = "*"
urllib3 = {extras = ["secure"],version = "*"}
facebook-sdk = "*"
[requires]
python_version = "3.8"
@ -55,6 +56,8 @@ pylint = "*"
pylint-django = "*"
unittest-xml-reporting = "*"
black = "*"
selenium = "*"
docker = "*"
[pipenv]
allow_prereleases = true

144
Pipfile.lock generated
View File

@ -1,7 +1,7 @@
{
"_meta": {
"hash": {
"sha256": "541f26a45f249fb2e61a597af7be7dee51eb8b40aa1035ae4081a455168128cc"
"sha256": "fd0192b73c01aaffb90716ce7b6d4e5be9adb8788d3ebd58e54ccd6f85d9b71b"
},
"pipfile-spec": 6,
"requires": {
@ -306,6 +306,14 @@
],
"version": "==1.0.0"
},
"facebook-sdk": {
"hashes": [
"sha256:2e987b3e0f466a6f4ee77b935eb023dba1384134f004a2af21f1cfff7fe0806e",
"sha256:cabcd2e69ea3d9f042919c99b353df7aa1e2be86d040121f6e9f5e63c1cf0f8d"
],
"index": "pypi",
"version": "==3.1.0"
},
"future": {
"hashes": [
"sha256:b1bead90b70cf6ec3f0710ae53a525360fa360d306a86583adc6bf83a4db537d"
@ -871,6 +879,53 @@
"index": "pypi",
"version": "==0.6.0"
},
"certifi": {
"hashes": [
"sha256:5ad7e9a056d25ffa5082862e36f119f7f7cec6457fa07ee2f8c339814b80c9b1",
"sha256:9cd41137dc19af6a5e03b630eefe7d1f458d964d406342dd3edf625839b944cc"
],
"version": "==2020.4.5.2"
},
"cffi": {
"hashes": [
"sha256:001bf3242a1bb04d985d63e138230802c6c8d4db3668fb545fb5005ddf5bb5ff",
"sha256:00789914be39dffba161cfc5be31b55775de5ba2235fe49aa28c148236c4e06b",
"sha256:028a579fc9aed3af38f4892bdcc7390508adabc30c6af4a6e4f611b0c680e6ac",
"sha256:14491a910663bf9f13ddf2bc8f60562d6bc5315c1f09c704937ef17293fb85b0",
"sha256:1cae98a7054b5c9391eb3249b86e0e99ab1e02bb0cc0575da191aedadbdf4384",
"sha256:2089ed025da3919d2e75a4d963d008330c96751127dd6f73c8dc0c65041b4c26",
"sha256:2d384f4a127a15ba701207f7639d94106693b6cd64173d6c8988e2c25f3ac2b6",
"sha256:337d448e5a725bba2d8293c48d9353fc68d0e9e4088d62a9571def317797522b",
"sha256:399aed636c7d3749bbed55bc907c3288cb43c65c4389964ad5ff849b6370603e",
"sha256:3b911c2dbd4f423b4c4fcca138cadde747abdb20d196c4a48708b8a2d32b16dd",
"sha256:3d311bcc4a41408cf5854f06ef2c5cab88f9fded37a3b95936c9879c1640d4c2",
"sha256:62ae9af2d069ea2698bf536dcfe1e4eed9090211dbaafeeedf5cb6c41b352f66",
"sha256:66e41db66b47d0d8672d8ed2708ba91b2f2524ece3dee48b5dfb36be8c2f21dc",
"sha256:675686925a9fb403edba0114db74e741d8181683dcf216be697d208857e04ca8",
"sha256:7e63cbcf2429a8dbfe48dcc2322d5f2220b77b2e17b7ba023d6166d84655da55",
"sha256:8a6c688fefb4e1cd56feb6c511984a6c4f7ec7d2a1ff31a10254f3c817054ae4",
"sha256:8c0ffc886aea5df6a1762d0019e9cb05f825d0eec1f520c51be9d198701daee5",
"sha256:95cd16d3dee553f882540c1ffe331d085c9e629499ceadfbda4d4fde635f4b7d",
"sha256:99f748a7e71ff382613b4e1acc0ac83bf7ad167fb3802e35e90d9763daba4d78",
"sha256:b8c78301cefcf5fd914aad35d3c04c2b21ce8629b5e4f4e45ae6812e461910fa",
"sha256:c420917b188a5582a56d8b93bdd8e0f6eca08c84ff623a4c16e809152cd35793",
"sha256:c43866529f2f06fe0edc6246eb4faa34f03fe88b64a0a9a942561c8e22f4b71f",
"sha256:cab50b8c2250b46fe738c77dbd25ce017d5e6fb35d3407606e7a4180656a5a6a",
"sha256:cef128cb4d5e0b3493f058f10ce32365972c554572ff821e175dbc6f8ff6924f",
"sha256:cf16e3cf6c0a5fdd9bc10c21687e19d29ad1fe863372b5543deaec1039581a30",
"sha256:e56c744aa6ff427a607763346e4170629caf7e48ead6921745986db3692f987f",
"sha256:e577934fc5f8779c554639376beeaa5657d54349096ef24abe8c74c5d9c117c3",
"sha256:f2b0fa0c01d8a0c7483afd9f31d7ecf2d71760ca24499c8697aeb5ca37dc090c"
],
"version": "==1.14.0"
},
"chardet": {
"hashes": [
"sha256:84ab92ed1c4d4f16916e05906b6b75a6c0fb5db821cc65e70cbd64a3e2a5eaae",
"sha256:fc323ffcaeaed0e0a02bf4d117757b98aed530d9ed4531e3e15460124c106691"
],
"version": "==3.0.4"
},
"click": {
"hashes": [
"sha256:d2b5255c7c6349bc1bd1e59e08cd12acbbd63ce649f2588755783aa94dfb6b1a",
@ -923,6 +978,30 @@
"index": "pypi",
"version": "==5.1"
},
"cryptography": {
"hashes": [
"sha256:091d31c42f444c6f519485ed528d8b451d1a0c7bf30e8ca583a0cac44b8a0df6",
"sha256:18452582a3c85b96014b45686af264563e3e5d99d226589f057ace56196ec78b",
"sha256:1dfa985f62b137909496e7fc182dac687206d8d089dd03eaeb28ae16eec8e7d5",
"sha256:1e4014639d3d73fbc5ceff206049c5a9a849cefd106a49fa7aaaa25cc0ce35cf",
"sha256:22e91636a51170df0ae4dcbd250d318fd28c9f491c4e50b625a49964b24fe46e",
"sha256:3b3eba865ea2754738616f87292b7f29448aec342a7c720956f8083d252bf28b",
"sha256:651448cd2e3a6bc2bb76c3663785133c40d5e1a8c1a9c5429e4354201c6024ae",
"sha256:726086c17f94747cedbee6efa77e99ae170caebeb1116353c6cf0ab67ea6829b",
"sha256:844a76bc04472e5135b909da6aed84360f522ff5dfa47f93e3dd2a0b84a89fa0",
"sha256:88c881dd5a147e08d1bdcf2315c04972381d026cdb803325c03fe2b4a8ed858b",
"sha256:96c080ae7118c10fcbe6229ab43eb8b090fccd31a09ef55f83f690d1ef619a1d",
"sha256:a0c30272fb4ddda5f5ffc1089d7405b7a71b0b0f51993cb4e5dbb4590b2fc229",
"sha256:bb1f0281887d89617b4c68e8db9a2c42b9efebf2702a3c5bf70599421a8623e3",
"sha256:c447cf087cf2dbddc1add6987bbe2f767ed5317adb2d08af940db517dd704365",
"sha256:c4fd17d92e9d55b84707f4fd09992081ba872d1a0c610c109c18e062e06a2e55",
"sha256:d0d5aeaedd29be304848f1c5059074a740fa9f6f26b84c5b63e8b29e73dfc270",
"sha256:daf54a4b07d67ad437ff239c8a4080cfd1cc7213df57d33c97de7b4738048d5e",
"sha256:e993468c859d084d5579e2ebee101de8f5a27ce8e2159959b6673b418fd8c785",
"sha256:f118a95c7480f5be0df8afeb9a11bd199aa20afab7a96bcf20409b411a3a85f0"
],
"version": "==2.9.2"
},
"django": {
"hashes": [
"sha256:5052b34b34b3425233c682e0e11d658fd6efd587d11335a0203d827224ada8f2",
@ -939,6 +1018,14 @@
"index": "pypi",
"version": "==2.2"
},
"docker": {
"hashes": [
"sha256:380a20d38fbfaa872e96ee4d0d23ad9beb0f9ed57ff1c30653cbeb0c9c0964f2",
"sha256:672f51aead26d90d1cfce84a87e6f71fca401bbc2a6287be18603583620a28ba"
],
"index": "pypi",
"version": "==4.2.1"
},
"gitdb": {
"hashes": [
"sha256:91f36bfb1ab7949b3b40e23736db18231bf7593edada2ba5c3a174a7b23657ac",
@ -953,6 +1040,13 @@
],
"version": "==3.1.3"
},
"idna": {
"hashes": [
"sha256:7588d1c14ae4c77d74036e8c22ff447b26d0fde8f007354fd48a7814db15b7cb",
"sha256:a068a21ceac8a4d63dbfd964670474107f541babbd2250d61922f029858365fa"
],
"version": "==2.9"
},
"isort": {
"hashes": [
"sha256:54da7e92468955c4fceacd0c86bd0ec997b0e1ee80d97f67c35a78b719dccab1",
@ -1014,6 +1108,13 @@
],
"version": "==2.6.0"
},
"pycparser": {
"hashes": [
"sha256:2d475327684562c3a96cc71adf7dc8c4f0565175cf86b6d7a404ff4c771f15f0",
"sha256:7582ad22678f0fcd81102833f60ef8d0e57288b6b5fb00323d101be910e35705"
],
"version": "==2.20"
},
"pylint": {
"hashes": [
"sha256:7dd78437f2d8d019717dbf287772d0b2dbdfd13fc016aa7faa08d67bccc46adc",
@ -1037,6 +1138,13 @@
],
"version": "==0.6"
},
"pyopenssl": {
"hashes": [
"sha256:621880965a720b8ece2f1b2f54ea2071966ab00e2970ad2ce11d596102063504",
"sha256:9a24494b2602aaf402be5c9e30a0b82d4a5c67528fe8fb475e3f3bc00dd69507"
],
"version": "==19.1.0"
},
"pytz": {
"hashes": [
"sha256:a494d53b6d39c3c6e44c3bec237336e14305e4f29bbf800b599253057fbb79ed",
@ -1087,6 +1195,21 @@
],
"version": "==2020.6.8"
},
"requests": {
"hashes": [
"sha256:b3559a131db72c33ee969480840fff4bb6dd111de7dd27c8ee1f820f4f00231b",
"sha256:fe75cc94a9443b9246fc7049224f75604b113c36acb93f87b80ed42c44cbb898"
],
"version": "==2.24.0"
},
"selenium": {
"hashes": [
"sha256:5f5489a0c5fe2f09cc6bc3f32a0d53441ab36882c987269f2afe805979633ac1",
"sha256:a9779ddc69cf03b75d94062c5e948f763919cf3341c77272f94cd05e6b4c7b32"
],
"index": "pypi",
"version": "==4.0.0a6.post2"
},
"six": {
"hashes": [
"sha256:30639c035cdb23534cd4aa2dd52c3bf48f06e5f4a941509c8bafd8ce11080259",
@ -1156,6 +1279,25 @@
"index": "pypi",
"version": "==3.0.2"
},
"urllib3": {
"extras": [
"secure"
],
"hashes": [
"sha256:3018294ebefce6572a474f0604c2021e33b3fd8006ecd11d62107a5d2a963527",
"sha256:88206b0eb87e6d677d424843ac5209e3fb9d0190d0ee169599165ec25e9d9115"
],
"index": "pypi",
"markers": null,
"version": "==1.25.9"
},
"websocket-client": {
"hashes": [
"sha256:0fc45c961324d79c781bab301359d5a1b00b13ad1b10415a4780229ef71a5549",
"sha256:d735b91d6d1692a6a181f2a8c9e0238e5f6373356f561bb9dc4c7af36f452010"
],
"version": "==0.57.0"
},
"wrapt": {
"hashes": [
"sha256:b62ffa81fb85f4332a4f609cab4ac40709470da05643a082ec1eb88e6d9b97d7"

9
docker.env.yml Normal file
View File

@ -0,0 +1,9 @@
debug: true
postgresql:
user: postgres
host: postgresql
redis:
host: redis
log_level: debug

0
e2e/__init__.py Normal file
View File

20
e2e/docker-compose.yml Normal file
View File

@ -0,0 +1,20 @@
version: '3.7'
services:
chrome:
image: selenium/standalone-chrome-debug:3.141.59-20200525
volumes:
- /dev/shm:/dev/shm
network_mode: host
postgresql:
image: postgres:11
restart: always
environment:
POSTGRES_HOST_AUTH_METHOD: trust
POSTGRES_DB: passbook
network_mode: host
redis:
image: redis
restart: always
network_mode: host

300
e2e/passbook.side Normal file
View File

@ -0,0 +1,300 @@
{
"id": "7d9b2407-1520-4c04-b040-68e8ada9aecc",
"version": "2.0",
"name": "passbook",
"url": "http://localhost:8000",
"tests": [{
"id": "94b39863-74ec-4b7d-98c5-2b380b6d2c55",
"name": "passbook login simple",
"commands": [{
"id": "e60e4382-4f96-44c3-ba06-5e18609c9c2b",
"comment": "",
"command": "open",
"target": "/flows/default-authentication-flow/?next=%2F",
"targets": [],
"value": ""
}, {
"id": "b2652f24-931e-45b0-b01d-2f0ac0f74db8",
"comment": "",
"command": "click",
"target": "id=id_uid_field",
"targets": [
["id=id_uid_field", "id"],
["name=uid_field", "name"],
["css=#id_uid_field", "css:finder"],
["xpath=//input[@id='id_uid_field']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div/input", "xpath:idRelative"],
["xpath=//div/input", "xpath:position"]
],
"value": ""
}, {
"id": "f1930f8a-984a-4076-a925-20937bb2f8d3",
"comment": "",
"command": "type",
"target": "id=id_uid_field",
"targets": [
["id=id_uid_field", "id"],
["name=uid_field", "name"],
["css=#id_uid_field", "css:finder"],
["xpath=//input[@id='id_uid_field']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div/input", "xpath:idRelative"],
["xpath=//div/input", "xpath:position"]
],
"value": "admin@example.tld"
}, {
"id": "0b568ee3-1bed-4821-a3bc-f6b960dbed9d",
"comment": "",
"command": "sendKeys",
"target": "id=id_uid_field",
"targets": [
["id=id_uid_field", "id"],
["name=uid_field", "name"],
["css=#id_uid_field", "css:finder"],
["xpath=//input[@id='id_uid_field']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div/input", "xpath:idRelative"],
["xpath=//div/input", "xpath:position"]
],
"value": "${KEY_ENTER}"
}, {
"id": "6d98e479-2825-484d-996a-ccf350d2761f",
"comment": "",
"command": "type",
"target": "id=id_password",
"targets": [
["id=id_password", "id"],
["name=password", "name"],
["css=#id_password", "css:finder"],
["xpath=//input[@id='id_password']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div[2]/input", "xpath:idRelative"],
["xpath=//div[2]/input", "xpath:position"]
],
"value": "pbadmin"
}, {
"id": "6f7abec6-ff44-4eb5-ae23-520c1c29a706",
"comment": "",
"command": "sendKeys",
"target": "id=id_password",
"targets": [
["id=id_password", "id"],
["name=password", "name"],
["css=#id_password", "css:finder"],
["xpath=//input[@id='id_password']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div[2]/input", "xpath:idRelative"],
["xpath=//div[2]/input", "xpath:position"]
],
"value": "${KEY_ENTER}"
}, {
"id": "04c5876f-1405-4077-a98b-e911f09113d7",
"comment": "",
"command": "assertText",
"target": "xpath=//a[contains(@href, '/-/user/')]",
"targets": [
["linkText=pbadmin", "linkText"],
["css=.pf-c-page__header-tools-group:nth-child(2) > .pf-c-button", "css:finder"],
["xpath=//a[contains(text(),'pbadmin')]", "xpath:link"],
["xpath=//div[@id='page-default-nav-example']/header/div[3]/div[2]/a", "xpath:idRelative"],
["xpath=//a[contains(@href, '/-/user/')]", "xpath:href"],
["xpath=//div[2]/a", "xpath:position"],
["xpath=//a[contains(.,'pbadmin')]", "xpath:innerText"]
],
"value": "pbadmin"
}]
}, {
"id": "61948b3c-3012-4f97-aa52-bc8f34fec333",
"name": "passbook enroll simple",
"commands": [{
"id": "0f4884b3-4891-41bc-956d-1fa433e892e9",
"comment": "",
"command": "open",
"target": "/flows/default-authentication-flow/?next=%2F",
"targets": [],
"value": ""
}, {
"id": "84d3861f-a60c-4650-8689-535f82b39577",
"comment": "",
"command": "click",
"target": "linkText=Sign up.",
"targets": [
["linkText=Sign up.", "linkText"],
["css=.pf-c-login__main-footer-band-item > a", "css:finder"],
["xpath=//a[contains(text(),'Sign up.')]", "xpath:link"],
["xpath=//main[@id='flow-body']/footer/div/p/a", "xpath:idRelative"],
["xpath=//a[contains(@href, '/flows/default-enrollment-flow/')]", "xpath:href"],
["xpath=//a", "xpath:position"],
["xpath=//a[contains(.,'Sign up.')]", "xpath:innerText"]
],
"value": ""
}, {
"id": "a32435ca-d84a-41e7-a915-fcbbc5f88341",
"comment": "",
"command": "type",
"target": "id=id_username",
"targets": [
["id=id_username", "id"],
["name=username", "name"],
["css=#id_username", "css:finder"],
["xpath=//input[@id='id_username']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div/input", "xpath:idRelative"],
["xpath=//div/input", "xpath:position"]
],
"value": "foo"
}, {
"id": "3b5dcf53-8297-46c5-88b7-11c2eb25f34f",
"comment": "",
"command": "type",
"target": "id=id_password",
"targets": [
["id=id_password", "id"],
["name=password", "name"],
["css=#id_password", "css:finder"],
["xpath=//input[@id='id_password']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div[2]/input", "xpath:idRelative"],
["xpath=//div[2]/input", "xpath:position"]
],
"value": "pbadmin"
}, {
"id": "e948d61c-dae6-4994-b56f-ff130892b342",
"comment": "",
"command": "type",
"target": "id=id_password_repeat",
"targets": [
["id=id_password_repeat", "id"],
["name=password_repeat", "name"],
["css=#id_password_repeat", "css:finder"],
["xpath=//input[@id='id_password_repeat']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div[3]/input", "xpath:idRelative"],
["xpath=//div[3]/input", "xpath:position"]
],
"value": "pbadmin"
}, {
"id": "e7527bfc-ec74-4d96-86f0-5a3a55a59025",
"comment": "",
"command": "click",
"target": "css=.pf-c-button",
"targets": [
["css=.pf-c-button", "css:finder"],
["xpath=//button[@type='submit']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div[4]/button", "xpath:idRelative"],
["xpath=//button", "xpath:position"],
["xpath=//button[contains(.,'Continue')]", "xpath:innerText"]
],
"value": ""
}, {
"id": "434b842c-a659-4ff5-aca8-06a6a3489597",
"comment": "",
"command": "type",
"target": "id=id_name",
"targets": [
["id=id_name", "id"],
["name=name", "name"],
["css=#id_name", "css:finder"],
["xpath=//input[@id='id_name']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div/input", "xpath:idRelative"],
["xpath=//div/input", "xpath:position"]
],
"value": "some name"
}, {
"id": "cbc43a1b-2cfe-46e2-85bc-476fb32c6cb1",
"comment": "",
"command": "type",
"target": "id=id_email",
"targets": [
["id=id_email", "id"],
["name=email", "name"],
["css=#id_email", "css:finder"],
["xpath=//input[@id='id_email']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div[2]/input", "xpath:idRelative"],
["xpath=//div[2]/input", "xpath:position"]
],
"value": "foo@bar.baz"
}, {
"id": "e74389a0-228b-4312-9677-e9add6358de3",
"comment": "",
"command": "click",
"target": "css=.pf-c-button",
"targets": [
["css=.pf-c-button", "css:finder"],
["xpath=//button[@type='submit']", "xpath:attributes"],
["xpath=//main[@id='flow-body']/div/form/div[3]/button", "xpath:idRelative"],
["xpath=//button", "xpath:position"],
["xpath=//button[contains(.,'Continue')]", "xpath:innerText"]
],
"value": ""
}, {
"id": "3e22f9c2-5ebd-49c2-81b1-340fa0435bbc",
"comment": "",
"command": "click",
"target": "linkText=foo",
"targets": [
["linkText=foo", "linkText"],
["css=.pf-c-page__header-tools-group:nth-child(2) > .pf-c-button", "css:finder"],
["xpath=//a[contains(text(),'foo')]", "xpath:link"],
["xpath=//div[@id='page-default-nav-example']/header/div[3]/div[2]/a", "xpath:idRelative"],
["xpath=//a[contains(@href, '/-/user/')]", "xpath:href"],
["xpath=//div[2]/a", "xpath:position"],
["xpath=//a[contains(.,'foo')]", "xpath:innerText"]
],
"value": ""
}, {
"id": "60124cfd-f11c-4d7f-8b01-bef54c8cbd73",
"comment": "",
"command": "assertText",
"target": "xpath=//a[contains(@href, '/-/user/')]",
"targets": [
["linkText=foo", "linkText"],
["css=.pf-c-page__header-tools-group:nth-child(2) > .pf-c-button", "css:finder"],
["xpath=//a[contains(text(),'foo')]", "xpath:link"],
["xpath=//div[@id='page-default-nav-example']/header/div[3]/div[2]/a", "xpath:idRelative"],
["xpath=//a[contains(@href, '/-/user/')]", "xpath:href"],
["xpath=//div[2]/a", "xpath:position"],
["xpath=//a[contains(.,'foo')]", "xpath:innerText"]
],
"value": "foo"
}, {
"id": "429ee61b-9991-4919-8131-55f8e1bd9a0d",
"comment": "",
"command": "assertValue",
"target": "id=id_username",
"targets": [],
"value": "foo"
}, {
"id": "f6c50760-52ed-4c1d-b232-30f8afe144eb",
"comment": "",
"command": "assertText",
"target": "id=id_name",
"targets": [
["id=id_name", "id"],
["name=name", "name"],
["css=#id_name", "css:finder"],
["xpath=//input[@id='id_name']", "xpath:attributes"],
["xpath=//main[@id='main-content']/section/div/div/div/div[2]/form/div[2]/div/input", "xpath:idRelative"],
["xpath=//div[2]/div/input", "xpath:position"]
],
"value": "some name"
}, {
"id": "b26905b5-89b5-4b41-abf5-a9f848f08622",
"comment": "",
"command": "assertText",
"target": "id=id_email",
"targets": [
["id=id_email", "id"],
["name=email", "name"],
["css=#id_email", "css:finder"],
["xpath=//input[@id='id_email']", "xpath:attributes"],
["xpath=//main[@id='main-content']/section/div/div/div/div[2]/form/div[3]/div/input", "xpath:idRelative"],
["xpath=//div[3]/div/input", "xpath:position"]
],
"value": "foo@bar.baz"
}]
}],
"suites": [{
"id": "495657fb-3f5e-4431-877c-4d0b248c0841",
"name": "Default Suite",
"persistSession": false,
"parallel": false,
"timeout": 300,
"tests": ["94b39863-74ec-4b7d-98c5-2b380b6d2c55"]
}],
"urls": ["http://localhost:8000/"],
"plugins": []
}

20
e2e/setup.sh Executable file
View File

@ -0,0 +1,20 @@
#!/bin/bash -x
# Setup docker & compose
curl -fsSL https://get.docker.com | bash
sudo usermod -a -G docker ubuntu
sudo curl -L "https://github.com/docker/compose/releases/download/1.26.0/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose
# Setup nodejs
curl -sL https://deb.nodesource.com/setup_12.x | sudo -E bash -
sudo apt-get install -y nodejs
sudo npm install -g yarn
# Setup python
sudo apt install -y python3.8 python3-pip
# Setup docker
sudo pip3 install pipenv
cd e2e
sudo docker-compose up -d
cd ..
pipenv sync --dev
pipenv shell

476
e2e/test_enroll.py Normal file
View File

@ -0,0 +1,476 @@
"""Test Enroll flow"""
from time import sleep
from django.test import override_settings
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as ec
from docker import DockerClient, from_env
from docker.models.containers import Container
from docker.types import Healthcheck
from e2e.utils import USER, SeleniumTestCase
from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
from passbook.policies.expression.models import ExpressionPolicy
from passbook.policies.models import PolicyBinding
from passbook.stages.email.models import EmailStage, EmailTemplates
from passbook.stages.identification.models import IdentificationStage
from passbook.stages.prompt.models import FieldTypes, Prompt, PromptStage
from passbook.stages.user_login.models import UserLoginStage
from passbook.stages.user_write.models import UserWriteStage
class TestEnroll(SeleniumTestCase):
"""Test Enroll flow"""
def setUp(self):
super().setUp()
self.container = self.setup_client()
def setup_client(self) -> Container:
"""Setup test IdP container"""
client: DockerClient = from_env()
container = client.containers.run(
image="mailhog/mailhog",
detach=True,
network_mode="host",
auto_remove=True,
healthcheck=Healthcheck(
test=["CMD", "wget", "-s", "http://localhost:8025"],
interval=5 * 100 * 1000000,
start_period=1 * 100 * 1000000,
),
)
while True:
container.reload()
status = container.attrs.get("State", {}).get("Health", {}).get("Status")
if status == "healthy":
return container
sleep(1)
def tearDown(self):
self.container.kill()
super().tearDown()
# pylint: disable=too-many-statements
def setup_test_enroll_2_step(self):
"""Setup all required objects"""
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.driver.find_element(By.LINK_TEXT, "Administrate").click()
self.driver.find_element(By.LINK_TEXT, "Prompts").click()
# Create Password Prompt
self.driver.find_element(By.LINK_TEXT, "Create").click()
self.driver.find_element(By.ID, "id_field_key").send_keys("password")
self.driver.find_element(By.ID, "id_label").send_keys("Password")
dropdown = self.driver.find_element(By.ID, "id_type")
dropdown.find_element(By.XPATH, "//option[. = 'Password']").click()
self.driver.find_element(By.ID, "id_placeholder").send_keys("Password")
self.driver.find_element(By.ID, "id_order").send_keys("1")
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
# Create Password Repeat Prompt
self.driver.find_element(By.LINK_TEXT, "Create").click()
self.driver.find_element(By.ID, "id_field_key").send_keys("password_repeat")
self.driver.find_element(By.ID, "id_label").send_keys("Password (repeat)")
dropdown = self.driver.find_element(By.ID, "id_type")
dropdown.find_element(By.XPATH, "//option[. = 'Password']").click()
self.driver.find_element(By.ID, "id_placeholder").send_keys("Password (repeat)")
self.driver.find_element(By.ID, "id_order").send_keys("2")
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
# Create Name Prompt
self.driver.find_element(By.LINK_TEXT, "Create").click()
self.driver.find_element(By.ID, "id_field_key").send_keys("name")
self.driver.find_element(By.ID, "id_label").send_keys("Name")
dropdown = self.driver.find_element(By.ID, "id_type")
dropdown.find_element(By.XPATH, "//option[. = 'Text']").click()
self.driver.find_element(By.ID, "id_placeholder").send_keys("Name")
self.driver.find_element(By.ID, "id_order").send_keys("0")
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
# Create Email Prompt
self.driver.find_element(By.LINK_TEXT, "Create").click()
self.driver.find_element(By.ID, "id_field_key").send_keys("email")
self.driver.find_element(By.ID, "id_label").send_keys("Email")
dropdown = self.driver.find_element(By.ID, "id_type")
dropdown.find_element(By.XPATH, "//option[. = 'Email']").click()
self.driver.find_element(By.ID, "id_placeholder").send_keys("Email")
self.driver.find_element(By.ID, "id_order").send_keys("1")
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
self.driver.find_element(By.LINK_TEXT, "Stages").click()
# Create first enroll prompt stage
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-dropdown__toggle").click()
self.driver.find_element(
By.CSS_SELECTOR, "li:nth-child(9) > .pf-c-dropdown__menu-item > small"
).click()
self.driver.find_element(By.ID, "id_name").send_keys(
"enroll-prompt-stage-first"
)
dropdown = self.driver.find_element(By.ID, "id_fields")
dropdown.find_element(
By.XPATH, "//option[. = \"Prompt 'username' type=text\"]"
).click()
dropdown.find_element(
By.XPATH, "//option[. = \"Prompt 'password' type=password\"]"
).click()
dropdown.find_element(
By.XPATH, "//option[. = \"Prompt 'password_repeat' type=password\"]"
).click()
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
# Create second enroll prompt stage
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-dropdown__toggle").click()
self.driver.find_element(
By.CSS_SELECTOR, "li:nth-child(9) > .pf-c-dropdown__menu-item"
).click()
self.driver.find_element(By.ID, "id_name").send_keys(
"enroll-prompt-stage-second"
)
dropdown = self.driver.find_element(By.ID, "id_fields")
dropdown.find_element(
By.XPATH, "//option[. = \"Prompt 'name' type=text\"]"
).click()
dropdown.find_element(
By.XPATH, "//option[. = \"Prompt 'email' type=email\"]"
).click()
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
# Create user write stage
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-dropdown__toggle").click()
self.driver.find_element(
By.CSS_SELECTOR, "li:nth-child(13) > .pf-c-dropdown__menu-item"
).click()
self.driver.find_element(By.ID, "id_name").send_keys("enroll-user-write")
self.driver.find_element(By.ID, "id_name").send_keys(Keys.ENTER)
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-dropdown__toggle").click()
# Create user login stage
self.driver.find_element(
By.CSS_SELECTOR, "li:nth-child(11) > .pf-c-dropdown__menu-item"
).click()
self.driver.find_element(By.ID, "id_name").send_keys("enroll-user-login")
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
self.driver.find_element(
By.CSS_SELECTOR,
".pf-c-nav__item:nth-child(7) .pf-c-nav__item:nth-child(1) > .pf-c-nav__link",
).click()
# Create password policy
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-dropdown__toggle").click()
self.driver.find_element(
By.CSS_SELECTOR, "li:nth-child(2) > .pf-c-dropdown__menu-item > small"
).click()
self.driver.find_element(By.ID, "id_name").send_keys(
"policy-enrollment-password-equals"
)
self.wait.until(
ec.presence_of_element_located((By.CSS_SELECTOR, ".CodeMirror-scroll"))
)
self.driver.find_element(By.CSS_SELECTOR, ".CodeMirror-scroll").click()
self.driver.find_element(By.CSS_SELECTOR, ".CodeMirror textarea").send_keys(
"return request.context['password'] == request.context['password_repeat']"
)
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
# Create password policy binding
self.driver.find_element(
By.CSS_SELECTOR,
".pf-c-nav__item:nth-child(7) .pf-c-nav__item:nth-child(2) > .pf-c-nav__link",
).click()
self.driver.find_element(By.LINK_TEXT, "Create").click()
dropdown = self.driver.find_element(By.ID, "id_policy")
dropdown.find_element(
By.XPATH, '//option[. = "Policy policy-enrollment-password-equals"]'
).click()
self.driver.find_element(By.ID, "id_target").click()
dropdown = self.driver.find_element(By.ID, "id_target")
dropdown.find_element(
By.XPATH, '//option[. = "Prompt Stage enroll-prompt-stage-first"]'
).click()
self.driver.find_element(By.ID, "id_order").send_keys("0")
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
# Create Flow
self.driver.find_element(
By.CSS_SELECTOR,
".pf-c-nav__item:nth-child(6) .pf-c-nav__item:nth-child(1) > .pf-c-nav__link",
).click()
self.driver.find_element(By.LINK_TEXT, "Create").click()
self.driver.find_element(By.ID, "id_name").send_keys("Welcome")
self.driver.find_element(By.ID, "id_slug").clear()
self.driver.find_element(By.ID, "id_slug").send_keys("default-enrollment-flow")
dropdown = self.driver.find_element(By.ID, "id_designation")
dropdown.find_element(By.XPATH, '//option[. = "Enrollment"]').click()
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
self.driver.find_element(By.LINK_TEXT, "Stages").click()
# Edit identification stage
self.driver.find_element(
By.CSS_SELECTOR, "tr:nth-child(11) .pf-m-secondary"
).click()
self.driver.find_element(
By.CSS_SELECTOR,
".pf-c-form__group:nth-child(5) .pf-c-form__horizontal-group",
).click()
self.driver.find_element(By.ID, "id_enrollment_flow").click()
dropdown = self.driver.find_element(By.ID, "id_enrollment_flow")
dropdown.find_element(
By.XPATH, '//option[. = "Flow Welcome (default-enrollment-flow)"]'
).click()
self.driver.find_element(By.ID, "id_user_fields_add_all_link").click()
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
self.driver.find_element(By.LINK_TEXT, "Bindings").click()
# Create Stage binding for first prompt stage
self.driver.find_element(By.LINK_TEXT, "Create").click()
self.driver.find_element(By.ID, "id_flow").click()
dropdown = self.driver.find_element(By.ID, "id_flow")
dropdown.find_element(
By.XPATH, '//option[. = "Flow Welcome (default-enrollment-flow)"]'
).click()
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-form").click()
self.driver.find_element(By.ID, "id_stage").click()
dropdown = self.driver.find_element(By.ID, "id_stage")
dropdown.find_element(
By.XPATH, '//option[. = "Stage enroll-prompt-stage-first"]'
).click()
self.driver.find_element(By.ID, "id_order").click()
self.driver.find_element(By.ID, "id_order").send_keys("0")
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
# Create Stage binding for second prompt stage
self.driver.find_element(By.LINK_TEXT, "Create").click()
self.driver.find_element(By.ID, "id_flow").click()
dropdown = self.driver.find_element(By.ID, "id_flow")
dropdown.find_element(
By.XPATH, '//option[. = "Flow Welcome (default-enrollment-flow)"]'
).click()
self.driver.find_element(By.ID, "id_stage").click()
dropdown = self.driver.find_element(By.ID, "id_stage")
dropdown.find_element(
By.XPATH, '//option[. = "Stage enroll-prompt-stage-second"]'
).click()
self.driver.find_element(By.ID, "id_order").click()
self.driver.find_element(By.ID, "id_order").send_keys("1")
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
# Create Stage binding for user write stage
self.driver.find_element(By.LINK_TEXT, "Create").click()
self.driver.find_element(By.ID, "id_flow").click()
dropdown = self.driver.find_element(By.ID, "id_flow")
dropdown.find_element(
By.XPATH, '//option[. = "Flow Welcome (default-enrollment-flow)"]'
).click()
self.driver.find_element(By.ID, "id_stage").click()
dropdown = self.driver.find_element(By.ID, "id_stage")
dropdown.find_element(
By.XPATH, '//option[. = "Stage enroll-user-write"]'
).click()
self.driver.find_element(By.ID, "id_order").click()
self.driver.find_element(By.ID, "id_order").send_keys("2")
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
# Create Stage binding for user login stage
self.driver.find_element(By.LINK_TEXT, "Create").click()
dropdown = self.driver.find_element(By.ID, "id_flow")
dropdown.find_element(
By.XPATH, '//option[. = "Flow Welcome (default-enrollment-flow)"]'
).click()
dropdown = self.driver.find_element(By.ID, "id_stage")
dropdown.find_element(
By.XPATH, '//option[. = "Stage enroll-user-login"]'
).click()
self.driver.find_element(By.ID, "id_order").send_keys("3")
self.driver.find_element(
By.CSS_SELECTOR, ".pf-c-form__actions > .pf-m-primary"
).click()
self.driver.find_element(By.CSS_SELECTOR, "[aria-label=logout]").click()
def test_enroll_2_step(self):
"""Test 2-step enroll flow"""
self.driver.get(self.live_server_url)
self.setup_test_enroll_2_step()
self.wait.until(
ec.presence_of_element_located((By.CSS_SELECTOR, "[role=enroll]"))
)
self.driver.find_element(By.CSS_SELECTOR, "[role=enroll]").click()
self.wait.until(ec.presence_of_element_located((By.ID, "id_username")))
self.driver.find_element(By.ID, "id_username").send_keys("foo")
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password_repeat").send_keys(USER().username)
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-button").click()
self.driver.find_element(By.ID, "id_name").send_keys("some name")
self.driver.find_element(By.ID, "id_email").send_keys("foo@bar.baz")
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-button").click()
self.wait.until(ec.presence_of_element_located((By.LINK_TEXT, "foo")))
self.driver.find_element(By.LINK_TEXT, "foo").click()
self.assertEqual(
self.driver.find_element(By.XPATH, "//a[contains(@href, '/-/user/')]").text,
"foo",
)
self.assertEqual(
self.driver.find_element(By.ID, "id_username").get_attribute("value"), "foo"
)
self.assertEqual(
self.driver.find_element(By.ID, "id_name").get_attribute("value"),
"some name",
)
self.assertEqual(
self.driver.find_element(By.ID, "id_email").get_attribute("value"),
"foo@bar.baz",
)
@override_settings(EMAIL_BACKEND="django.core.mail.backends.smtp.EmailBackend")
def test_enroll_email(self):
"""Test enroll with Email verification"""
# First stage fields
username_prompt = Prompt.objects.create(
field_key="username", label="Username", order=0, type=FieldTypes.TEXT
)
password = Prompt.objects.create(
field_key="password", label="Password", order=1, type=FieldTypes.PASSWORD
)
password_repeat = Prompt.objects.create(
field_key="password_repeat",
label="Password (repeat)",
order=2,
type=FieldTypes.PASSWORD,
)
# Second stage fields
name_field = Prompt.objects.create(
field_key="name", label="Name", order=0, type=FieldTypes.TEXT
)
email = Prompt.objects.create(
field_key="email", label="E-Mail", order=1, type=FieldTypes.EMAIL
)
# Stages
first_stage = PromptStage.objects.create(name="prompt-stage-first")
first_stage.fields.set([username_prompt, password, password_repeat])
first_stage.save()
second_stage = PromptStage.objects.create(name="prompt-stage-second")
second_stage.fields.set([name_field, email])
second_stage.save()
email_stage = EmailStage.objects.create(
name="enroll-email",
host="localhost",
port=1025,
template=EmailTemplates.ACCOUNT_CONFIRM,
)
user_write = UserWriteStage.objects.create(name="enroll-user-write")
user_login = UserLoginStage.objects.create(name="enroll-user-login")
# Password checking policy
password_policy = ExpressionPolicy.objects.create(
name="policy-enrollment-password-equals",
expression="return request.context['password'] == request.context['password_repeat']",
)
PolicyBinding.objects.create(
target=first_stage, policy=password_policy, order=0
)
flow = Flow.objects.create(
name="default-enrollment-flow",
slug="default-enrollment-flow",
designation=FlowDesignation.ENROLLMENT,
)
# Attach enrollment flow to identification stage
ident_stage: IdentificationStage = IdentificationStage.objects.first()
ident_stage.enrollment_flow = flow
ident_stage.save()
FlowStageBinding.objects.create(flow=flow, stage=first_stage, order=0)
FlowStageBinding.objects.create(flow=flow, stage=second_stage, order=1)
FlowStageBinding.objects.create(flow=flow, stage=user_write, order=2)
FlowStageBinding.objects.create(flow=flow, stage=email_stage, order=3)
FlowStageBinding.objects.create(flow=flow, stage=user_login, order=4)
self.driver.get(self.live_server_url)
self.driver.find_element(By.CSS_SELECTOR, "[role=enroll]").click()
self.driver.find_element(By.ID, "id_username").send_keys("foo")
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password_repeat").send_keys(USER().username)
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-button").click()
self.driver.find_element(By.ID, "id_name").send_keys("some name")
self.driver.find_element(By.ID, "id_email").send_keys("foo@bar.baz")
self.driver.find_element(By.CSS_SELECTOR, ".pf-c-button").click()
sleep(3)
# Open Mailhog
self.driver.get("http://localhost:8025")
# Click on first message
self.driver.find_element(By.CLASS_NAME, "msglist-message").click()
sleep(3)
self.driver.switch_to.frame(self.driver.find_element(By.CLASS_NAME, "tab-pane"))
self.driver.find_element(By.ID, "confirm").click()
self.driver.close()
self.driver.switch_to.window(self.driver.window_handles[0])
# We're now logged in
sleep(3)
self.wait.until(
ec.presence_of_element_located(
(By.XPATH, "//a[contains(@href, '/-/user/')]")
)
)
self.driver.find_element(By.XPATH, "//a[contains(@href, '/-/user/')]").click()
self.assertEqual(
self.driver.find_element(By.XPATH, "//a[contains(@href, '/-/user/')]").text,
"foo",
)
self.assertEqual(
self.driver.find_element(By.ID, "id_username").get_attribute("value"), "foo"
)
self.assertEqual(
self.driver.find_element(By.ID, "id_name").get_attribute("value"),
"some name",
)
self.assertEqual(
self.driver.find_element(By.ID, "id_email").get_attribute("value"),
"foo@bar.baz",
)

22
e2e/test_login_default.py Normal file
View File

@ -0,0 +1,22 @@
"""test default login flow"""
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from e2e.utils import USER, SeleniumTestCase
class TestLogin(SeleniumTestCase):
"""test default login flow"""
def test_login(self):
"""test default login flow"""
self.driver.get(f"{self.live_server_url}/flows/default-authentication-flow/")
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.assertEqual(
self.driver.find_element(By.XPATH, "//a[contains(@href, '/-/user/')]").text,
USER().username,
)

191
e2e/test_provider_oauth.py Normal file
View File

@ -0,0 +1,191 @@
"""test OAuth Provider flow"""
from time import sleep
from oauth2_provider.generators import generate_client_id, generate_client_secret
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from docker import DockerClient, from_env
from docker.models.containers import Container
from docker.types import Healthcheck
from e2e.utils import USER, SeleniumTestCase
from passbook.core.models import Application
from passbook.flows.models import Flow
from passbook.providers.oauth.models import OAuth2Provider
class TestProviderOAuth(SeleniumTestCase):
"""test OAuth Provider flow"""
def setUp(self):
super().setUp()
self.client_id = generate_client_id()
self.client_secret = generate_client_secret()
self.container = self.setup_client()
def setup_client(self) -> Container:
"""Setup client grafana container which we test OAuth against"""
client: DockerClient = from_env()
container = client.containers.run(
image="grafana/grafana:latest",
detach=True,
network_mode="host",
auto_remove=True,
healthcheck=Healthcheck(
test=["CMD", "wget", "--spider", "http://localhost:3000"],
interval=5 * 100 * 1000000,
start_period=1 * 100 * 1000000,
),
environment={
"GF_AUTH_GITHUB_ENABLED": "true",
"GF_AUTH_GITHUB_allow_sign_up": "true",
"GF_AUTH_GITHUB_CLIENT_ID": self.client_id,
"GF_AUTH_GITHUB_CLIENT_SECRET": self.client_secret,
"GF_AUTH_GITHUB_SCOPES": "user:email,read:org",
"GF_AUTH_GITHUB_AUTH_URL": self.url(
"passbook_providers_oauth:github-authorize"
),
"GF_AUTH_GITHUB_TOKEN_URL": self.url(
"passbook_providers_oauth:github-access-token"
),
"GF_AUTH_GITHUB_API_URL": self.url(
"passbook_providers_oauth:github-user"
),
"GF_LOG_LEVEL": "debug",
},
)
while True:
container.reload()
status = container.attrs.get("State", {}).get("Health", {}).get("Status")
if status == "healthy":
return container
sleep(1)
def tearDown(self):
self.container.kill()
super().tearDown()
def test_authorization_consent_implied(self):
"""test OAuth Provider flow (default authorization flow with implied consent)"""
sleep(1)
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-implicit-consent"
)
provider = OAuth2Provider.objects.create(
name="grafana",
client_type=OAuth2Provider.CLIENT_CONFIDENTIAL,
authorization_grant_type=OAuth2Provider.GRANT_AUTHORIZATION_CODE,
client_id=self.client_id,
client_secret=self.client_secret,
redirect_uris="http://localhost:3000/login/github",
skip_authorization=True,
authorization_flow=authorization_flow,
)
Application.objects.create(
name="Grafana", slug="grafana", provider=provider,
)
self.driver.get("http://localhost:3000")
self.driver.find_element(By.CLASS_NAME, "btn-service--github").click()
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
USER().username,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[1]/div/input",
).get_attribute("value"),
USER().username,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[2]/div/input",
).get_attribute("value"),
USER().email,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[3]/div/input",
).get_attribute("value"),
USER().username,
)
def test_authorization_consent_explicit(self):
"""test OAuth Provider flow (default authorization flow with explicit consent)"""
sleep(1)
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-explicit-consent"
)
provider = OAuth2Provider.objects.create(
name="grafana",
client_type=OAuth2Provider.CLIENT_CONFIDENTIAL,
authorization_grant_type=OAuth2Provider.GRANT_AUTHORIZATION_CODE,
client_id=self.client_id,
client_secret=self.client_secret,
redirect_uris="http://localhost:3000/login/github",
skip_authorization=True,
authorization_flow=authorization_flow,
)
app = Application.objects.create(
name="Grafana", slug="grafana", provider=provider,
)
self.driver.get("http://localhost:3000")
self.driver.find_element(By.CLASS_NAME, "btn-service--github").click()
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.assertIn(
app.name,
self.driver.find_element(
By.XPATH, "/html/body/div[2]/div/main/div/form/div[2]/p[1]"
).text,
)
self.assertEqual(
"GitHub Compatibility: User Email",
self.driver.find_element(
By.XPATH, "/html/body/div[2]/div/main/div/form/div[2]/ul/li[1]"
).text,
)
self.driver.find_element(By.CSS_SELECTOR, "[type=submit]").click()
self.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
USER().username,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[1]/div/input",
).get_attribute("value"),
USER().username,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[2]/div/input",
).get_attribute("value"),
USER().email,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[3]/div/input",
).get_attribute("value"),
USER().username,
)

254
e2e/test_provider_oidc.py Normal file
View File

@ -0,0 +1,254 @@
"""test OpenID Provider flow"""
from time import sleep
from django.shortcuts import reverse
from oauth2_provider.generators import generate_client_id, generate_client_secret
from oidc_provider.models import Client, ResponseType
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as ec
from docker import DockerClient, from_env
from docker.models.containers import Container
from docker.types import Healthcheck
from e2e.utils import USER, SeleniumTestCase, ensure_rsa_key
from passbook.core.models import Application
from passbook.flows.models import Flow
from passbook.providers.oidc.models import OpenIDProvider
class TestProviderOIDC(SeleniumTestCase):
"""test OpenID Provider flow"""
def setUp(self):
super().setUp()
self.client_id = generate_client_id()
self.client_secret = generate_client_secret()
self.container = self.setup_client()
def setup_client(self) -> Container:
"""Setup client grafana container which we test OIDC against"""
client: DockerClient = from_env()
container = client.containers.run(
image="grafana/grafana:latest",
detach=True,
network_mode="host",
auto_remove=True,
healthcheck=Healthcheck(
test=["CMD", "wget", "--spider", "http://localhost:3000"],
interval=5 * 100 * 1000000,
start_period=1 * 100 * 1000000,
),
environment={
"GF_AUTH_GENERIC_OAUTH_ENABLED": "true",
"GF_AUTH_GENERIC_OAUTH_CLIENT_ID": self.client_id,
"GF_AUTH_GENERIC_OAUTH_CLIENT_SECRET": self.client_secret,
"GF_AUTH_GENERIC_OAUTH_SCOPES": "openid email profile",
"GF_AUTH_GENERIC_OAUTH_AUTH_URL": (
self.live_server_url + reverse("passbook_providers_oidc:authorize")
),
"GF_AUTH_GENERIC_OAUTH_TOKEN_URL": (
self.live_server_url + reverse("oidc_provider:token")
),
"GF_AUTH_GENERIC_OAUTH_API_URL": (
self.live_server_url + reverse("oidc_provider:userinfo")
),
"GF_LOG_LEVEL": "debug",
},
)
while True:
container.reload()
status = container.attrs.get("State", {}).get("Health", {}).get("Status")
if status == "healthy":
return container
sleep(1)
def tearDown(self):
self.container.kill()
super().tearDown()
def test_redirect_uri_error(self):
"""test OpenID Provider flow (invalid redirect URI, check error message)"""
sleep(1)
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-implicit-consent"
)
client = Client.objects.create(
name="grafana",
client_type="confidential",
client_id=self.client_id,
client_secret=self.client_secret,
_redirect_uris="http://localhost:3000/",
_scope="openid userinfo",
)
# At least one of these objects must exist
ensure_rsa_key()
# This response_code object might exist or not, depending on the order the tests are run
rp_type, _ = ResponseType.objects.get_or_create(value="code")
client.response_types.set([rp_type])
client.save()
provider = OpenIDProvider.objects.create(
oidc_client=client, authorization_flow=authorization_flow,
)
Application.objects.create(
name="Grafana", slug="grafana", provider=provider,
)
self.driver.get("http://localhost:3000")
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
sleep(2)
self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "pf-c-title").text,
"Redirect URI Error",
)
def test_authorization_consent_implied(self):
"""test OpenID Provider flow (default authorization flow with implied consent)"""
sleep(1)
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-implicit-consent"
)
client = Client.objects.create(
name="grafana",
client_type="confidential",
client_id=self.client_id,
client_secret=self.client_secret,
_redirect_uris="http://localhost:3000/login/generic_oauth",
_scope="openid profile email",
reuse_consent=False,
require_consent=False,
)
# At least one of these objects must exist
ensure_rsa_key()
# This response_code object might exist or not, depending on the order the tests are run
rp_type, _ = ResponseType.objects.get_or_create(value="code")
client.response_types.set([rp_type])
client.save()
provider = OpenIDProvider.objects.create(
oidc_client=client, authorization_flow=authorization_flow,
)
Application.objects.create(
name="Grafana", slug="grafana", provider=provider,
)
self.driver.get("http://localhost:3000")
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
USER().name,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[1]/div/input",
).get_attribute("value"),
USER().name,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[2]/div/input",
).get_attribute("value"),
USER().email,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[3]/div/input",
).get_attribute("value"),
USER().email,
)
def test_authorization_consent_explicit(self):
"""test OpenID Provider flow (default authorization flow with explicit consent)"""
sleep(1)
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-explicit-consent"
)
client = Client.objects.create(
name="grafana",
client_type="confidential",
client_id=self.client_id,
client_secret=self.client_secret,
_redirect_uris="http://localhost:3000/login/generic_oauth",
_scope="openid profile email",
reuse_consent=False,
require_consent=False,
)
# At least one of these objects must exist
ensure_rsa_key()
# This response_code object might exist or not, depending on the order the tests are run
rp_type, _ = ResponseType.objects.get_or_create(value="code")
client.response_types.set([rp_type])
client.save()
provider = OpenIDProvider.objects.create(
oidc_client=client, authorization_flow=authorization_flow,
)
app = Application.objects.create(
name="Grafana", slug="grafana", provider=provider,
)
self.driver.get("http://localhost:3000")
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.assertIn(
app.name,
self.driver.find_element(
By.XPATH, "/html/body/div[2]/div/main/div/form/div[2]/p[1]"
).text,
)
self.wait.until(
ec.presence_of_element_located((By.CSS_SELECTOR, "[type=submit]"))
)
self.driver.find_element(By.CSS_SELECTOR, "[type=submit]").click()
self.wait.until(
ec.presence_of_element_located(
(By.XPATH, "//a[contains(@href, '/profile')]")
)
)
self.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
USER().name,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[1]/div/input",
).get_attribute("value"),
USER().name,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[2]/div/input",
).get_attribute("value"),
USER().email,
)
self.assertEqual(
self.driver.find_element(
By.XPATH,
"/html/body/grafana-app/div/div/div/react-profile-wrapper/form[1]/div[3]/div/input",
).get_attribute("value"),
USER().email,
)

172
e2e/test_provider_saml.py Normal file
View File

@ -0,0 +1,172 @@
"""test SAML Provider flow"""
from time import sleep
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from docker import DockerClient, from_env
from docker.models.containers import Container
from docker.types import Healthcheck
from e2e.utils import USER, SeleniumTestCase
from passbook.core.models import Application
from passbook.crypto.models import CertificateKeyPair
from passbook.flows.models import Flow
from passbook.lib.utils.reflection import class_to_path
from passbook.providers.saml.models import (
SAMLBindings,
SAMLPropertyMapping,
SAMLProvider,
)
from passbook.providers.saml.processors.generic import GenericProcessor
class TestProviderSAML(SeleniumTestCase):
"""test SAML Provider flow"""
container: Container
def setup_client(self, provider: SAMLProvider) -> Container:
"""Setup client saml-sp container which we test SAML against"""
client: DockerClient = from_env()
container = client.containers.run(
image="beryju/saml-test-sp",
detach=True,
network_mode="host",
auto_remove=True,
healthcheck=Healthcheck(
test=["CMD", "wget", "--spider", "http://localhost:9009/health"],
interval=5 * 100 * 1000000,
start_period=1 * 100 * 1000000,
),
environment={
"SP_ENTITY_ID": provider.issuer,
"SP_SSO_BINDING": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST",
"SP_METADATA_URL": (
self.url(
"passbook_providers_saml:metadata",
application_slug=provider.application.slug,
)
),
},
)
while True:
container.reload()
status = container.attrs.get("State", {}).get("Health", {}).get("Status")
if status == "healthy":
return container
sleep(1)
def tearDown(self):
self.container.kill()
super().tearDown()
def test_sp_initiated_implicit(self):
"""test SAML Provider flow SP-initiated flow (implicit consent)"""
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-implicit-consent"
)
provider: SAMLProvider = SAMLProvider.objects.create(
name="saml-test",
processor_path=class_to_path(GenericProcessor),
acs_url="http://localhost:9009/saml/acs",
audience="passbook-e2e",
issuer="passbook-e2e",
sp_binding=SAMLBindings.POST,
authorization_flow=authorization_flow,
signing_kp=CertificateKeyPair.objects.first(),
)
provider.property_mappings.set(SAMLPropertyMapping.objects.all())
provider.save()
Application.objects.create(
name="SAML", slug="passbook-saml", provider=provider,
)
self.container = self.setup_client(provider)
self.driver.get("http://localhost:9009")
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.assertEqual(
self.driver.find_element(By.XPATH, "/html/body/pre").text,
f"Hello, {USER().name}!",
)
def test_sp_initiated_explicit(self):
"""test SAML Provider flow SP-initiated flow (explicit consent)"""
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-explicit-consent"
)
provider: SAMLProvider = SAMLProvider.objects.create(
name="saml-test",
processor_path=class_to_path(GenericProcessor),
acs_url="http://localhost:9009/saml/acs",
audience="passbook-e2e",
issuer="passbook-e2e",
sp_binding=SAMLBindings.POST,
authorization_flow=authorization_flow,
signing_kp=CertificateKeyPair.objects.first(),
)
provider.property_mappings.set(SAMLPropertyMapping.objects.all())
provider.save()
app = Application.objects.create(
name="SAML", slug="passbook-saml", provider=provider,
)
self.container = self.setup_client(provider)
self.driver.get("http://localhost:9009")
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.assertIn(
app.name,
self.driver.find_element(
By.XPATH, "/html/body/div[2]/div/main/div/form/div[2]/p[1]"
).text,
)
self.driver.find_element(By.CSS_SELECTOR, "[type=submit]").click()
self.assertEqual(
self.driver.find_element(By.XPATH, "/html/body/pre").text,
f"Hello, {USER().name}!",
)
def test_idp_initiated_implicit(self):
"""test SAML Provider flow IdP-initiated flow (implicit consent)"""
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-implicit-consent"
)
provider: SAMLProvider = SAMLProvider.objects.create(
name="saml-test",
processor_path=class_to_path(GenericProcessor),
acs_url="http://localhost:9009/saml/acs",
audience="passbook-e2e",
issuer="passbook-e2e",
sp_binding=SAMLBindings.POST,
authorization_flow=authorization_flow,
signing_kp=CertificateKeyPair.objects.first(),
)
provider.property_mappings.set(SAMLPropertyMapping.objects.all())
provider.save()
Application.objects.create(
name="SAML", slug="passbook-saml", provider=provider,
)
self.container = self.setup_client(provider)
self.driver.get(
self.url(
"passbook_providers_saml:sso-init",
application_slug=provider.application.slug,
)
)
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.assertEqual(
self.driver.find_element(By.XPATH, "/html/body/pre").text,
f"Hello, {USER().name}!",
)

127
e2e/test_source_saml.py Normal file
View File

@ -0,0 +1,127 @@
"""test SAML Source"""
from time import sleep
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as ec
from docker import DockerClient, from_env
from docker.models.containers import Container
from docker.types import Healthcheck
from e2e.utils import SeleniumTestCase
from passbook.crypto.models import CertificateKeyPair
from passbook.flows.models import Flow
from passbook.sources.saml.models import SAMLBindingTypes, SAMLSource
IDP_CERT = """-----BEGIN CERTIFICATE-----
MIIDXTCCAkWgAwIBAgIJALmVVuDWu4NYMA0GCSqGSIb3DQEBCwUAMEUxCzAJBgNV
BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX
aWRnaXRzIFB0eSBMdGQwHhcNMTYxMjMxMTQzNDQ3WhcNNDgwNjI1MTQzNDQ3WjBF
MQswCQYDVQQGEwJBVTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50
ZXJuZXQgV2lkZ2l0cyBQdHkgTHRkMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB
CgKCAQEAzUCFozgNb1h1M0jzNRSCjhOBnR+uVbVpaWfXYIR+AhWDdEe5ryY+Cgav
Og8bfLybyzFdehlYdDRgkedEB/GjG8aJw06l0qF4jDOAw0kEygWCu2mcH7XOxRt+
YAH3TVHa/Hu1W3WjzkobqqqLQ8gkKWWM27fOgAZ6GieaJBN6VBSMMcPey3HWLBmc
+TYJmv1dbaO2jHhKh8pfKw0W12VM8P1PIO8gv4Phu/uuJYieBWKixBEyy0lHjyix
YFCR12xdh4CA47q958ZRGnnDUGFVE1QhgRacJCOZ9bd5t9mr8KLaVBYTCJo5ERE8
jymab5dPqe5qKfJsCZiqWglbjUo9twIDAQABo1AwTjAdBgNVHQ4EFgQUxpuwcs/C
YQOyui+r1G+3KxBNhxkwHwYDVR0jBBgwFoAUxpuwcs/CYQOyui+r1G+3KxBNhxkw
DAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAAiWUKs/2x/viNCKi3Y6b
lEuCtAGhzOOZ9EjrvJ8+COH3Rag3tVBWrcBZ3/uhhPq5gy9lqw4OkvEws99/5jFs
X1FJ6MKBgqfuy7yh5s1YfM0ANHYczMmYpZeAcQf2CGAaVfwTTfSlzNLsF2lW/ly7
yapFzlYSJLGoVE+OHEu8g5SlNACUEfkXw+5Eghh+KzlIN7R6Q7r2ixWNFBC/jWf7
NKUfJyX8qIG5md1YUeT6GBW9Bm2/1/RiO24JTaYlfLdKK9TYb8sG5B+OLab2DImG
99CJ25RkAcSobWNF5zD0O6lgOo3cEdB/ksCq3hmtlC/DlLZ/D8CJ+7VuZnS1rR2n
aQ==
-----END CERTIFICATE-----"""
class TestSourceSAML(SeleniumTestCase):
"""test SAML Source flow"""
def setUp(self):
super().setUp()
self.container = self.setup_client()
def setup_client(self) -> Container:
"""Setup test IdP container"""
client: DockerClient = from_env()
container = client.containers.run(
image="kristophjunge/test-saml-idp",
detach=True,
network_mode="host",
auto_remove=True,
healthcheck=Healthcheck(
test=["CMD", "curl", "http://localhost:8080"],
interval=5 * 100 * 1000000,
start_period=1 * 100 * 1000000,
),
environment={
"SIMPLESAMLPHP_SP_ENTITY_ID": "entity-id",
"SIMPLESAMLPHP_SP_ASSERTION_CONSUMER_SERVICE": (
f"{self.live_server_url}/source/saml/saml-idp-test/acs/"
),
},
)
while True:
container.reload()
status = container.attrs.get("State", {}).get("Health", {}).get("Status")
if status == "healthy":
return container
sleep(1)
def tearDown(self):
self.container.kill()
super().tearDown()
def test_idp_redirect(self):
"""test SAML Source With redirect binding"""
sleep(1)
# Bootstrap all needed objects
authentication_flow = Flow.objects.get(slug="default-source-authentication")
enrollment_flow = Flow.objects.get(slug="default-source-enrollment")
keypair = CertificateKeyPair.objects.create(
name="test-idp-cert", certificate_data=IDP_CERT
)
SAMLSource.objects.create(
name="saml-idp-test",
slug="saml-idp-test",
authentication_flow=authentication_flow,
enrollment_flow=enrollment_flow,
issuer="entity-id",
sso_url="http://localhost:8080/simplesaml/saml2/idp/SSOService.php",
binding_type=SAMLBindingTypes.Redirect,
signing_kp=keypair,
)
self.driver.get(self.live_server_url)
self.wait.until(
ec.presence_of_element_located(
(By.CLASS_NAME, "pf-c-login__main-footer-links-item-link")
)
)
self.driver.find_element(
By.CLASS_NAME, "pf-c-login__main-footer-links-item-link"
).click()
# Now we should be at the IDP, wait for the username field
self.wait.until(ec.presence_of_element_located((By.ID, "username")))
self.driver.find_element(By.ID, "username").send_keys("user1")
self.driver.find_element(By.ID, "password").send_keys("user1pass")
self.driver.find_element(By.ID, "password").send_keys(Keys.ENTER)
# Wait until we're logged in
self.wait.until(
ec.presence_of_element_located(
(By.XPATH, "//a[contains(@href, '/-/user/')]")
)
)
self.driver.find_element(By.XPATH, "//a[contains(@href, '/-/user/')]").click()
# Wait until we've loaded the user info page
self.wait.until(ec.presence_of_element_located((By.ID, "id_username")))
self.assertNotEqual(
self.driver.find_element(By.ID, "id_username").get_attribute("value"), ""
)

92
e2e/utils.py Normal file
View File

@ -0,0 +1,92 @@
"""passbook e2e testing utilities"""
from functools import lru_cache
from glob import glob
from importlib.util import module_from_spec, spec_from_file_location
from inspect import getmembers, isfunction
from os import makedirs
from time import time
from Cryptodome.PublicKey import RSA
from django.apps import apps
from django.contrib.staticfiles.testing import StaticLiveServerTestCase
from django.db import connection, transaction
from django.db.utils import IntegrityError
from django.shortcuts import reverse
from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
from selenium.webdriver.remote.webdriver import WebDriver
from selenium.webdriver.support.ui import WebDriverWait
from passbook.core.models import User
@lru_cache
# pylint: disable=invalid-name
def USER() -> User: # noqa
"""Cached function that always returns pbadmin"""
return User.objects.get(username="pbadmin")
def ensure_rsa_key():
"""Ensure that at least one RSAKey Object exists, create one if none exist"""
from oidc_provider.models import RSAKey
if not RSAKey.objects.exists():
key = RSA.generate(2048)
rsakey = RSAKey(key=key.exportKey("PEM").decode("utf8"))
rsakey.save()
class SeleniumTestCase(StaticLiveServerTestCase):
"""StaticLiveServerTestCase which automatically creates a Webdriver instance"""
def setUp(self):
super().setUp()
makedirs("out", exist_ok=True)
self.driver = self._get_driver()
self.driver.maximize_window()
self.driver.implicitly_wait(5)
self.wait = WebDriverWait(self.driver, 60)
self.apply_default_data()
def _get_driver(self) -> WebDriver:
return webdriver.Remote(
command_executor="http://localhost:4444/wd/hub",
desired_capabilities=DesiredCapabilities.CHROME,
)
def tearDown(self):
self.driver.save_screenshot(f"out/{self.__class__.__name__}_{time()}.png")
self.driver.quit()
super().tearDown()
def url(self, view, **kwargs) -> str:
"""reverse `view` with `**kwargs` into full URL using live_server_url"""
return self.live_server_url + reverse(view, kwargs=kwargs)
def apply_default_data(self):
"""apply objects created by migrations after tables have been truncated"""
# Find all migration files
# load all functions
migration_files = glob("**/migrations/*.py", recursive=True)
matches = []
for migration in migration_files:
with open(migration, "r+") as migration_file:
# Check if they have a `RunPython`
if "RunPython" in migration_file.read():
matches.append(migration)
with connection.schema_editor() as schema_editor:
for match in matches:
# Load module from file path
spec = spec_from_file_location("", match)
migration_module = module_from_spec(spec)
# pyright: reportGeneralTypeIssues=false
spec.loader.exec_module(migration_module)
# Call all functions from module
for _, func in getmembers(migration_module, isfunction):
with transaction.atomic():
try:
func(apps, schema_editor)
except IntegrityError:
pass

1
manage.py
View File

@ -2,6 +2,7 @@
"""Django manage.py"""
import os
import sys
from defusedxml import defuse_stdlib
defuse_stdlib()

8
passbook/admin/templates/administration/stage_binding/list.html
View File

@ -39,8 +39,8 @@
{% for flow in grouped_bindings %}
<tr role="role">
<td>
{% blocktrans with name=flow.grouper.name %}
Flow {{ name }}
{% blocktrans with slug=flow.grouper.slug %}
Flow {{ slug }}
{% endblocktrans %}
</td>
<td></td>
@ -56,9 +56,9 @@
</td>
<th role="columnheader">
<div>
<div>{{ binding.flow.name }}</div>
<div>{{ binding.flow.slug }}</div>
<small>
{{ binding.flow }}
{{ binding.flow.name }}
</small>
</div>
</th>

2
passbook/api/v2/urls.py
View File

@ -30,6 +30,7 @@ from passbook.providers.oidc.api import OpenIDProviderViewSet
from passbook.providers.saml.api import SAMLPropertyMappingViewSet, SAMLProviderViewSet
from passbook.sources.ldap.api import LDAPPropertyMappingViewSet, LDAPSourceViewSet
from passbook.sources.oauth.api import OAuthSourceViewSet
from passbook.sources.saml.api import SAMLSourceViewSet
from passbook.stages.captcha.api import CaptchaStageViewSet
from passbook.stages.dummy.api import DummyStageViewSet
from passbook.stages.email.api import EmailStageViewSet
@ -61,6 +62,7 @@ router.register("audit/events", EventViewSet)
router.register("sources/all", SourceViewSet)
router.register("sources/ldap", LDAPSourceViewSet)
router.register("sources/saml", SAMLSourceViewSet)
router.register("sources/oauth", OAuthSourceViewSet)
router.register("policies/all", PolicyViewSet)

4
passbook/core/migrations/0003_default_user.py
View File

@ -9,7 +9,9 @@ def create_default_user(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
# We have to use a direct import here, otherwise we get an object manager error
from passbook.core.models import User
pbadmin, _ = User.objects.get_or_create(
db_alias = schema_editor.connection.alias
pbadmin, _ = User.objects.using(db_alias).get_or_create(
username="pbadmin", email="root@localhost", name="passbook Default Admin"
)
pbadmin.set_password("pbadmin") # noqa # nosec

2
passbook/core/templates/base/page.html
View File

@ -40,7 +40,7 @@
</div>
<div class="pf-c-page__header-tools">
<div class="pf-c-page__header-tools-group pf-m-icons">
<a href="{% url 'passbook_flows:default-invalidation' %}" class="pf-c-button pf-m-plain" type="button">
<a href="{% url 'passbook_flows:default-invalidation' %}" class="pf-c-button pf-m-plain" type="button" aria-label="logout">
<i class="fas fa-sign-out-alt" aria-hidden="true"></i>
</a>
</div>

48
passbook/core/templates/error/generic.html
View File

@ -1,15 +1,39 @@
{% extends 'login/base.html' %}
{% extends 'base/skeleton.html' %}
{% load static %}
{% load i18n %}
{% load passbook_utils %}
{% block card_title %}
{% block body %}
<div class="pf-c-background-image">
<svg xmlns="http://www.w3.org/2000/svg" class="pf-c-background-image__filter" width="0" height="0">
<filter id="image_overlay">
<feColorMatrix type="matrix" values="1 0 0 0 0 1 0 0 0 0 1 0 0 0 0 0 0 0 1 0"></feColorMatrix>
<feComponentTransfer color-interpolation-filters="sRGB" result="duotone">
<feFuncR type="table" tableValues="0.086274509803922 0.43921568627451"></feFuncR>
<feFuncG type="table" tableValues="0.086274509803922 0.43921568627451"></feFuncG>
<feFuncB type="table" tableValues="0.086274509803922 0.43921568627451"></feFuncB>
<feFuncA type="table" tableValues="0 1"></feFuncA>
</feComponentTransfer>
</filter>
</svg>
</div>
<div class="pf-c-login">
<div class="pf-c-login__container">
<header class="pf-c-login__header">
<img class="pf-c-brand" src="{% static 'passbook/logo.svg' %}" style="height: 60px;" alt="passbook icon" />
<img class="pf-c-brand" src="{% static 'passbook/brand.svg' %}" style="height: 60px;"
alt="passbook branding" />
</header>
<main class="pf-c-login__main" id="flow-body">
<header class="pf-c-login__main-header">
<h1 class="pf-c-title pf-m-3xl">
{% trans 'Bad Request' %}
{% endblock %}
</h1>
</header>
<div class="pf-c-login__main-body">
{% block card %}
<form>
<form method="POST" class="pf-c-form">
{% if message %}
<h3>{% trans message %}</h3>
{% endif %}
@ -18,3 +42,17 @@
{% endif %}
</form>
{% endblock %}
</div>
</main>
<footer class="pf-c-login__footer">
<p></p>
<ul class="pf-c-list pf-m-inline">
<li>
<a href="https://passbook.beryju.org/">{% trans 'Documentation' %}</a>
</li>
<!-- todo: load config.passbook.footer.links -->
</ul>
</footer>
</div>
</div>
{% endblock %}

6
passbook/flows/apps.py
View File

@ -1,4 +1,6 @@
"""passbook flows app config"""
from importlib import import_module
from django.apps import AppConfig
@ -9,3 +11,7 @@ class PassbookFlowsConfig(AppConfig):
label = "passbook_flows"
mountpoint = "flows/"
verbose_name = "passbook Flows"
def ready(self):
"""Load policy cache clearing signals"""
import_module("passbook.flows.signals")

67
passbook/flows/migrations/0004_source_flows.py
View File

@ -7,15 +7,12 @@ from django.db.backends.base.schema import BaseDatabaseSchemaEditor
from passbook.flows.models import FlowDesignation
from passbook.stages.prompt.models import FieldTypes
FLOW_POLICY_EXPRESSION = """{{ pb_is_sso_flow }}"""
PROMPT_POLICY_EXPRESSION = """
{% if pb_flow_plan.context.prompt_data.username %}
False
{% else %}
True
{% endif %}
"""
FLOW_POLICY_EXPRESSION = """# This policy ensures that this flow can only be used when the user
# is in a SSO Flow (meaning they come from an external IdP)
return pb_is_sso_flow"""
PROMPT_POLICY_EXPRESSION = """# Check if we've been given a username by the external IdP
# and trigger the enrollment flow
return 'username' in pb_flow_plan.context.get('prompt_data', {})"""
def create_default_source_enrollment_flow(
@ -37,25 +34,27 @@ def create_default_source_enrollment_flow(
db_alias = schema_editor.connection.alias
# Create a policy that only allows this flow when doing an SSO Request
flow_policy = ExpressionPolicy.objects.create(
flow_policy = ExpressionPolicy.objects.using(db_alias).create(
name="default-source-enrollment-if-sso", expression=FLOW_POLICY_EXPRESSION
)
# This creates a Flow used by sources to enroll users
# It makes sure that a username is set, and if not, prompts the user for a Username
flow = Flow.objects.create(
flow = Flow.objects.using(db_alias).create(
name="default-source-enrollment",
slug="default-source-enrollment",
designation=FlowDesignation.ENROLLMENT,
)
PolicyBinding.objects.create(policy=flow_policy, target=flow, order=0)
PolicyBinding.objects.using(db_alias).create(
policy=flow_policy, target=flow, order=0
)
# PromptStage to ask user for their username
prompt_stage = PromptStage.objects.create(
prompt_stage = PromptStage.objects.using(db_alias).create(
name="default-source-enrollment-username-prompt",
)
prompt_stage.fields.add(
Prompt.objects.create(
Prompt.objects.using(db_alias).create(
field_key="username",
label="Username",
type=FieldTypes.TEXT,
@ -64,20 +63,32 @@ def create_default_source_enrollment_flow(
)
)
# Policy to only trigger prompt when no username is given
prompt_policy = ExpressionPolicy.objects.create(
prompt_policy = ExpressionPolicy.objects.using(db_alias).create(
name="default-source-enrollment-if-username",
expression=PROMPT_POLICY_EXPRESSION,
)
# UserWrite stage to create the user, and login stage to log user in
user_write = UserWriteStage.objects.create(name="default-source-enrollment-write")
user_login = UserLoginStage.objects.create(name="default-source-enrollment-login")
user_write = UserWriteStage.objects.using(db_alias).create(
name="default-source-enrollment-write"
)
user_login = UserLoginStage.objects.using(db_alias).create(
name="default-source-enrollment-login"
)
binding = FlowStageBinding.objects.create(flow=flow, stage=prompt_stage, order=0)
PolicyBinding.objects.create(policy=prompt_policy, target=binding)
binding = FlowStageBinding.objects.using(db_alias).create(
flow=flow, stage=prompt_stage, order=0
)
PolicyBinding.objects.using(db_alias).create(
policy=prompt_policy, target=binding, order=0
)
FlowStageBinding.objects.create(flow=flow, stage=user_write, order=1)
FlowStageBinding.objects.create(flow=flow, stage=user_login, order=2)
FlowStageBinding.objects.using(db_alias).create(
flow=flow, stage=user_write, order=1
)
FlowStageBinding.objects.using(db_alias).create(
flow=flow, stage=user_login, order=2
)
def create_default_source_authentication_flow(
@ -96,22 +107,26 @@ def create_default_source_authentication_flow(
db_alias = schema_editor.connection.alias
# Create a policy that only allows this flow when doing an SSO Request
flow_policy = ExpressionPolicy.objects.create(
flow_policy = ExpressionPolicy.objects.using(db_alias).create(
name="default-source-authentication-if-sso", expression=FLOW_POLICY_EXPRESSION
)
# This creates a Flow used by sources to authenticate users
flow = Flow.objects.create(
flow = Flow.objects.using(db_alias).create(
name="default-source-authentication",
slug="default-source-authentication",
designation=FlowDesignation.AUTHENTICATION,
)
PolicyBinding.objects.create(policy=flow_policy, target=flow, order=0)
PolicyBinding.objects.using(db_alias).create(
policy=flow_policy, target=flow, order=0
)
user_login = UserLoginStage.objects.create(
user_login = UserLoginStage.objects.using(db_alias).create(
name="default-source-authentication-login"
)
FlowStageBinding.objects.create(flow=flow, stage=user_login, order=0)
FlowStageBinding.objects.using(db_alias).create(
flow=flow, stage=user_login, order=0
)
class Migration(migrations.Migration):

22
passbook/flows/migrations/0005_provider_flows.py
View File

@ -17,21 +17,23 @@ def create_default_provider_authz_flow(
db_alias = schema_editor.connection.alias
# Empty flow for providers where no consent is needed
Flow.objects.create(
name="default-provider-authorization",
slug="default-provider-authorization",
# Empty flow for providers where consent is implicitly given
Flow.objects.using(db_alias).create(
name="Authorize Application",
slug="default-provider-authorization-implicit-consent",
designation=FlowDesignation.AUTHORIZATION,
)
# Flow with consent form to obtain user consent for authorization
flow = Flow.objects.create(
name="default-provider-authorization-consent",
slug="default-provider-authorization-consent",
# Flow with consent form to obtain explicit user consent
flow = Flow.objects.using(db_alias).create(
name="Authorize Application",
slug="default-provider-authorization-explicit-consent",
designation=FlowDesignation.AUTHORIZATION,
)
stage = ConsentStage.objects.create(name="default-provider-authorization-consent")
FlowStageBinding.objects.create(flow=flow, stage=stage, order=0)
stage = ConsentStage.objects.using(db_alias).create(
name="default-provider-authorization-consent"
)
FlowStageBinding.objects.using(db_alias).create(flow=flow, stage=stage, order=0)
class Migration(migrations.Migration):

5
passbook/flows/planner.py
View File

@ -39,6 +39,11 @@ class FlowPlan:
context: Dict[str, Any] = field(default_factory=dict)
markers: List[StageMarker] = field(default_factory=list)
def append(self, stage: Stage, marker: Optional[StageMarker] = None):
"""Append `stage` to all stages, optionall with stage marker"""
self.stages.append(stage)
self.markers.append(marker or StageMarker())
def next(self) -> Optional[Stage]:
"""Return next pending stage from the bottom of the list"""
if not self.has_stages:

31
passbook/flows/signals.py Normal file
View File

@ -0,0 +1,31 @@
"""passbook flow signals"""
from django.core.cache import cache
from django.db.models.signals import post_save
from django.dispatch import receiver
from structlog import get_logger
LOGGER = get_logger()
@receiver(post_save)
# pylint: disable=unused-argument
def invalidate_flow_cache(sender, instance, **_):
"""Invalidate flow cache when flow is updated"""
from passbook.flows.models import Flow, FlowStageBinding, Stage
from passbook.flows.planner import cache_key
if isinstance(instance, Flow):
LOGGER.debug("Invalidating Flow cache", flow=instance)
cache.delete(f"{cache_key(instance)}*")
if isinstance(instance, FlowStageBinding):
LOGGER.debug("Invalidating Flow cache from FlowStageBinding", binding=instance)
cache.delete(f"{cache_key(instance.flow)}*")
if isinstance(instance, Stage):
LOGGER.debug("Invalidating Flow cache from Stage", stage=instance)
total = 0
for binding in FlowStageBinding.objects.filter(stage=instance):
prefix = cache_key(binding.flow)
keys = cache.keys(f"{prefix}*")
total += len(keys)
cache.delete_many(keys)
LOGGER.debug("Deleted keys", len=total)

7
passbook/flows/views.py
View File

@ -27,6 +27,7 @@ LOGGER = get_logger()
# Argument used to redirect user after login
NEXT_ARG_NAME = "next"
SESSION_KEY_PLAN = "passbook_flows_plan"
SESSION_KEY_GET = "passbook_flows_get"
@method_decorator(xframe_options_sameorigin, name="dispatch")
@ -127,7 +128,10 @@ class FlowExecutorView(View):
def _flow_done(self) -> HttpResponse:
"""User Successfully passed all stages"""
self.cancel()
next_param = self.request.GET.get(NEXT_ARG_NAME, "passbook_core:overview")
# Since this is wrapped by the ExecutorShell, the next argument is saved in the session
next_param = self.request.session.get(SESSION_KEY_GET, {}).get(
NEXT_ARG_NAME, "passbook_core:overview"
)
return redirect_with_qs(next_param)
def stage_ok(self) -> HttpResponse:
@ -210,6 +214,7 @@ class FlowExecutorShellView(TemplateView):
def get_context_data(self, **kwargs) -> Dict[str, Any]:
kwargs["exec_url"] = reverse("passbook_flows:flow-executor", kwargs=self.kwargs)
kwargs["msg_url"] = reverse("passbook_api:messages-list")
self.request.session[SESSION_KEY_GET] = self.request.GET
return kwargs

2
passbook/policies/apps.py
View File

@ -12,5 +12,5 @@ class PassbookPoliciesConfig(AppConfig):
verbose_name = "passbook Policies"
def ready(self):
"""Load source_types from config file"""
"""Load policy cache clearing signals"""
import_module("passbook.policies.signals")

2
passbook/policies/expression/evaluator.py
View File

@ -22,7 +22,7 @@ class PolicyEvaluator(BaseEvaluator):
super().__init__()
self._messages = []
self._context["pb_message"] = self.expr_func_message
self._filename = policy_name
self._filename = policy_name or "PolicyEvaluator"
def expr_func_message(self, message: str):
"""Wrapper to append to messages list, which is returned with PolicyResult"""

2
passbook/providers/oauth/models.py
View File

@ -23,7 +23,7 @@ class OAuth2Provider(Provider, AbstractApplication):
def html_setup_urls(self, request: HttpRequest) -> Optional[str]:
"""return template and context modal with URLs for authorize, token, openid-config, etc"""
return render_to_string(
"oauth2_provider/setup_url_modal.html",
"providers/oauth/setup_url_modal.html",
{
"provider": self,
"authorize_url": request.build_absolute_uri(

73
passbook/providers/oauth/templates/oauth2_provider/authorize.html
View File

@ -1,73 +0,0 @@
{% extends "login/base.html" %}
{% load passbook_utils %}
{% load i18n %}
{% block card_title %}
{% trans 'Authorize Application' %}
{% endblock %}
{% block card %}
<form method="POST" class="pf-c-form">
{% csrf_token %}
{% if not error %}
{% csrf_token %}
{% for field in form %}
{% if field.is_hidden %}
{{ field }}
{% endif %}
{% endfor %}
<div class="pf-c-form__group">
<p class="subtitle">
{% blocktrans with remote=application.name %}
You're about to sign into {{ remote }}.
{% endblocktrans %}
</p>
<p>{% trans "Application requires following permissions" %}</p>
<ul class="pf-c-list">
{% for scope in scopes_descriptions %}
<li>{{ scope }}</li>
{% endfor %}
</ul>
{{ form.errors }}
{{ form.non_field_errors }}
</div>
<div class="pf-c-form__group">
<p>
{% blocktrans with user=user %}
You are logged in as {{ user }}. Not you?
{% endblocktrans %}
<a href="{% url 'passbook_flows:default-invalidation' %}">{% trans 'Logout' %}</a>
</p>
</div>
<div class="pf-c-form__group pf-m-action">
<input type="submit" class="pf-c-button pf-m-primary" name="allow" value="{% trans 'Continue' %}">
<a href="{% back %}" class="pf-c-button pf-m-secondary">{% trans "Cancel" %}</a>
</div>
<div class="pf-c-form__group" style="display: none;" id="loading">
<div class="pf-c-form__horizontal-group">
<span class="pf-c-spinner" role="progressbar" aria-valuetext="Loading...">
<span class="pf-c-spinner__clipper"></span>
<span class="pf-c-spinner__lead-ball"></span>
<span class="pf-c-spinner__tail-ball"></span>
</span>
</div>
</div>
{% else %}
<div class="login-group">
<p class="subtitle">
{% blocktrans with err=error.error %}Error: {{ err }}{% endblocktrans %}
</p>
<p>{{ error.description }}</p>
</div>
{% endif %}
</form>
{% endblock %}
{% block scripts %}
<script>
document.querySelector("form").addEventListener("submit", (e) => {
document.getElementById("loading").removeAttribute("style");
});
</script>
{% endblock %}

1
passbook/providers/oauth/templates/oauth2_provider/base.html
View File

@ -1 +0,0 @@
{% extends "base/skeleton.html" %}

20
passbook/providers/oauth/templates/providers/oauth/consent.html Normal file
View File

@ -0,0 +1,20 @@
{% extends 'login/form_with_user.html' %}
{% load i18n %}
{% block beneath_form %}
<div class="pf-c-form__group">
<p>
{% blocktrans with name=context.application.name %}
You're about to sign into {{ name }}.
{% endblocktrans %}
</p>
<p>{% trans "Application requires following permissions" %}</p>
<ul class="pf-c-list">
{% for scope in context.scope_descriptions %}
<li>{{ scope }}</li>
{% endfor %}
</ul>
{{ hidden_inputs }}
</div>
{% endblock %}

0
passbook/providers/oauth/templates/oauth2_provider/setup_url_modal.html → passbook/providers/oauth/templates/providers/oauth/setup_url_modal.html
View File

20
passbook/providers/oauth/views/oauth2.py
View File

@ -1,9 +1,11 @@
"""passbook OAuth2 Views"""
from django.contrib import messages
from django.contrib.auth.mixins import LoginRequiredMixin
from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
from django.shortcuts import get_object_or_404, redirect
from django.views import View
from oauth2_provider.exceptions import OAuthToolkitError
from oauth2_provider.scopes import get_scopes_backend
from oauth2_provider.views.base import AuthorizationView
from structlog import get_logger
@ -20,6 +22,7 @@ from passbook.flows.stage import StageView
from passbook.flows.views import SESSION_KEY_PLAN
from passbook.lib.utils.urls import redirect_with_qs
from passbook.providers.oauth.models import OAuth2Provider
from passbook.stages.consent.stage import PLAN_CONTEXT_CONSENT_TEMPLATE
LOGGER = get_logger()
@ -32,9 +35,10 @@ PLAN_CONTEXT_CODE_CHALLENGE = "code_challenge"
PLAN_CONTEXT_CODE_CHALLENGE_METHOD = "code_challenge_method"
PLAN_CONTEXT_SCOPE = "scope"
PLAN_CONTEXT_NONCE = "nonce"
PLAN_CONTEXT_SCOPE_DESCRIPTION = "scope_descriptions"
class AuthorizationFlowInitView(AccessMixin, View):
class AuthorizationFlowInitView(AccessMixin, LoginRequiredMixin, View):
"""OAuth2 Flow initializer, checks access to application and starts flow"""
# pylint: disable=unused-argument
@ -54,8 +58,11 @@ class AuthorizationFlowInitView(AccessMixin, View):
return redirect("passbook_providers_oauth:oauth2-permission-denied")
# Regardless, we start the planner and return to it
planner = FlowPlanner(provider.authorization_flow)
# planner.use_cache = False
planner.allow_empty_flows = True
# Save scope descriptions
scopes = request.GET.get(PLAN_CONTEXT_SCOPE)
all_scopes = get_scopes_backend().get_all_scopes()
plan = planner.plan(
self.request,
{
@ -65,11 +72,16 @@ class AuthorizationFlowInitView(AccessMixin, View):
PLAN_CONTEXT_REDIRECT_URI: request.GET.get(PLAN_CONTEXT_REDIRECT_URI),
PLAN_CONTEXT_RESPONSE_TYPE: request.GET.get(PLAN_CONTEXT_RESPONSE_TYPE),
PLAN_CONTEXT_STATE: request.GET.get(PLAN_CONTEXT_STATE),
PLAN_CONTEXT_SCOPE: request.GET.get(PLAN_CONTEXT_SCOPE),
PLAN_CONTEXT_SCOPE: scopes,
PLAN_CONTEXT_NONCE: request.GET.get(PLAN_CONTEXT_NONCE),
PLAN_CONTEXT_SCOPE_DESCRIPTION: [
all_scopes[scope] for scope in scopes.split(" ")
],
PLAN_CONTEXT_CONSENT_TEMPLATE: "providers/oauth/consent.html",
},
)
plan.stages.append(in_memory_stage(OAuth2Stage))
plan.append(in_memory_stage(OAuth2Stage))
self.request.session[SESSION_KEY_PLAN] = plan
return redirect_with_qs(
"passbook_flows:flow-executor-shell",

2
passbook/providers/oidc/claims.py
View File

@ -10,5 +10,5 @@ def userinfo(claims: Dict[str, Any], user: User) -> Dict[str, Any]:
claims["given_name"] = user.name
claims["family_name"] = user.name
claims["email"] = user.email
claims["preferred_username"] = user.username
return claims

20
passbook/providers/oidc/templates/providers/oidc/consent.html Normal file
View File

@ -0,0 +1,20 @@
{% extends 'login/form_with_user.html' %}
{% load i18n %}
{% block beneath_form %}
<div class="pf-c-form__group">
<p>
{% blocktrans with name=context.application.name %}
You're about to sign into {{ name }}.
{% endblocktrans %}
</p>
<p>{% trans "Application requires following permissions" %}</p>
<ul class="pf-c-list">
{% for scope in context.scopes %}
<li>{{ scope.name }}</li>
{% endfor %}
</ul>
{{ hidden_inputs }}
</div>
{% endblock %}

9
passbook/providers/oidc/views.py
View File

@ -1,5 +1,6 @@
"""passbook OIDC Views"""
from django.contrib import messages
from django.contrib.auth.mixins import LoginRequiredMixin
from django.http import HttpRequest, HttpResponse, JsonResponse
from django.shortcuts import get_object_or_404, redirect, reverse
from django.views import View
@ -22,13 +23,15 @@ from passbook.flows.stage import StageView
from passbook.flows.views import SESSION_KEY_PLAN
from passbook.lib.utils.urls import redirect_with_qs
from passbook.providers.oidc.models import OpenIDProvider
from passbook.stages.consent.stage import PLAN_CONTEXT_CONSENT_TEMPLATE
LOGGER = get_logger()
PLAN_CONTEXT_PARAMS = "params"
PLAN_CONTEXT_SCOPES = "scopes"
class AuthorizationFlowInitView(AccessMixin, View):
class AuthorizationFlowInitView(AccessMixin, LoginRequiredMixin, View):
"""OIDC Flow initializer, checks access to application and starts flow"""
# pylint: disable=unused-argument
@ -58,9 +61,11 @@ class AuthorizationFlowInitView(AccessMixin, View):
PLAN_CONTEXT_SSO: True,
PLAN_CONTEXT_APPLICATION: application,
PLAN_CONTEXT_PARAMS: endpoint.params,
PLAN_CONTEXT_SCOPES: endpoint.get_scopes_information(),
PLAN_CONTEXT_CONSENT_TEMPLATE: "providers/oidc/consent.html",
},
)
plan.stages.append(in_memory_stage(OIDCStage))
plan.append(in_memory_stage(OIDCStage))
self.request.session[SESSION_KEY_PLAN] = plan
return redirect_with_qs(
"passbook_flows:flow-executor-shell",

2
passbook/providers/saml/forms.py
View File

@ -59,7 +59,7 @@ class SAMLProviderForm(forms.ModelForm):
class SAMLPropertyMappingForm(forms.ModelForm):
"""SAML Property Mapping form"""
template_name = "saml/idp/property_mapping_form.html"
template_name = "providers/saml/property_mapping_form.html"
def clean_expression(self):
"""Test Syntax"""

22
passbook/providers/saml/migrations/0004_auto_20200620_1950.py Normal file
View File

@ -0,0 +1,22 @@
# Generated by Django 3.0.7 on 2020-06-20 19:50
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("passbook_providers_saml", "0003_samlprovider_sp_binding"),
]
operations = [
migrations.AlterField(
model_name="samlprovider",
name="sp_binding",
field=models.TextField(
choices=[("redirect", "Redirect"), ("post", "Post")],
default="redirect",
verbose_name="Service Prodier Binding",
),
),
]

8
passbook/providers/saml/models.py
View File

@ -25,7 +25,7 @@ class SAMLBindings(models.TextChoices):
class SAMLProvider(Provider):
"""Model to save information about a Remote SAML Endpoint"""
"""SAML 2.0-based authentication protocol."""
name = models.TextField()
processor_path = models.CharField(max_length=255, choices=[])
@ -34,7 +34,9 @@ class SAMLProvider(Provider):
audience = models.TextField(default="")
issuer = models.TextField(help_text=_("Also known as EntityID"))
sp_binding = models.TextField(
choices=SAMLBindings.choices, default=SAMLBindings.REDIRECT
choices=SAMLBindings.choices,
default=SAMLBindings.REDIRECT,
verbose_name=_("Service Prodier Binding"),
)
assertion_valid_not_before = models.TextField(
@ -142,7 +144,7 @@ class SAMLProvider(Provider):
# pylint: disable=no-member
metadata = DescriptorDownloadView.get_metadata(request, self)
return render_to_string(
"saml/idp/admin_metadata_modal.html",
"providers/saml/admin_metadata_modal.html",
{"provider": self, "metadata": metadata},
)
except Provider.application.RelatedObjectDoesNotExist:

4
passbook/providers/saml/processors/base.py
View File

@ -132,7 +132,9 @@ class Processor:
continue
self._assertion_params["ATTRIBUTES"] = attributes
self._assertion_xml = get_assertion_xml(
"saml/xml/assertions/generic.xml", self._assertion_params, signed=True
"providers/saml/xml/assertions/generic.xml",
self._assertion_params,
signed=True,
)
def _format_response(self):

4
passbook/providers/saml/processors/salesforce.py
View File

@ -10,5 +10,7 @@ class SalesForceProcessor(GenericProcessor):
def _format_assertion(self):
super()._format_assertion()
self._assertion_xml = get_assertion_xml(
"saml/xml/assertions/salesforce.xml", self._assertion_params, signed=True
"providers/saml/xml/assertions/salesforce.xml",
self._assertion_params,
signed=True,
)

0
passbook/providers/saml/templates/saml/idp/admin_metadata_modal.html → passbook/providers/saml/templates/providers/saml/admin_metadata_modal.html
View File

0
passbook/providers/saml/templates/saml/idp/autosubmit_form.html → passbook/providers/saml/templates/providers/saml/autosubmit_form.html
View File

14
passbook/providers/saml/templates/providers/saml/consent.html Normal file
View File

@ -0,0 +1,14 @@
{% extends 'login/form_with_user.html' %}
{% load i18n %}
{% block beneath_form %}
<div class="pf-c-form__group">
<p>
{% blocktrans with name=context.application.name %}
You're about to sign into {{ name }}.
{% endblocktrans %}
</p>
{{ hidden_inputs }}
</div>
{% endblock %}

0
passbook/providers/saml/templates/saml/idp/logged_out.html → passbook/providers/saml/templates/providers/saml/logged_out.html
View File

0
passbook/providers/saml/templates/saml/idp/property_mapping_form.html → passbook/providers/saml/templates/providers/saml/property_mapping_form.html
View File

2
passbook/providers/saml/templates/saml/xml/assertions/generic.xml → passbook/providers/saml/templates/providers/saml/xml/assertions/generic.xml
View File

@ -3,7 +3,7 @@
IssueInstant="{{ ISSUE_INSTANT }}"
Version="2.0">
<saml:Issuer xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion">{{ ISSUER }}</saml:Issuer>
{% include 'saml/xml/signature.xml' %}
{% include 'providers/saml/xml/signature.xml' %}
{{ SUBJECT_STATEMENT }}
<saml:Conditions NotBefore="{{ NOT_BEFORE }}" NotOnOrAfter="{{ NOT_ON_OR_AFTER }}">
<saml:AudienceRestriction>

4
passbook/providers/saml/templates/saml/xml/assertions/google_apps.xml → passbook/providers/saml/templates/providers/saml/xml/assertions/google_apps.xml
View File

@ -3,8 +3,8 @@
IssueInstant="{{ ISSUE_INSTANT }}"
Version="2.0">
<saml:Issuer>{{ ISSUER }}</saml:Issuer>
{% include 'saml/xml/signature.xml' %}
{% include 'saml/xml/subject.xml' %}
{% include 'providers/saml/xml/signature.xml' %}
{% include 'providers/saml/xml/subject.xml' %}
<saml:Conditions NotBefore="{{ NOT_BEFORE }}" NotOnOrAfter="{{ NOT_ON_OR_AFTER }}" />
<saml:AuthnStatement AuthnInstant="{{ AUTH_INSTANT }}">
<saml:AuthnContext>

2
passbook/providers/saml/templates/saml/xml/assertions/salesforce.xml → passbook/providers/saml/templates/providers/saml/xml/assertions/salesforce.xml
View File

@ -4,7 +4,7 @@
Version="2.0">
<saml:Issuer xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion">{{ ISSUER }}</saml:Issuer>
{{ ASSERTION_SIGNATURE|safe }}
{% include 'saml/xml/subject.xml' %}
{% include 'providers/saml/xml/subject.xml' %}
<saml:Conditions NotBefore="{{ NOT_BEFORE }}" NotOnOrAfter="{{ NOT_ON_OR_AFTER }}">
<saml:AudienceRestriction>
<saml:Audience>{{ AUDIENCE }}</saml:Audience>

0
passbook/providers/saml/templates/saml/xml/attributes.xml → passbook/providers/saml/templates/providers/saml/xml/attributes.xml
View File

4
passbook/providers/saml/templates/saml/xml/metadata.xml → passbook/providers/saml/templates/providers/saml/xml/metadata.xml
View File

@ -11,7 +11,7 @@
</md:KeyDescriptor>
{% endif %}
<md:NameIDFormat>{{ subject_format }}</md:NameIDFormat>
<md:SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST" Location="{{ sso_binding_post }}"/>
<md:SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect" Location="{{ sso_binding_redirect }}"/>
<md:SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST" Location="{{ saml_sso_binding_post }}"/>
<md:SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect" Location="{{ saml_sso_binding_redirect }}"/>
</md:IDPSSODescriptor>
</md:EntityDescriptor>

0
passbook/providers/saml/templates/saml/xml/response.xml → passbook/providers/saml/templates/providers/saml/xml/response.xml
View File

0
passbook/providers/saml/templates/saml/xml/signature.xml → passbook/providers/saml/templates/providers/saml/xml/signature.xml
View File

0
passbook/providers/saml/templates/saml/xml/subject.xml → passbook/providers/saml/templates/providers/saml/xml/subject.xml
View File

8
passbook/providers/saml/utils/xml_render.py
View File

@ -28,7 +28,7 @@ def _get_attribute_statement(params):
return
# Build complete AttributeStatement.
params["ATTRIBUTE_STATEMENT"] = render_to_string(
"saml/xml/attributes.xml", {"attributes": attributes}
"providers/saml/xml/attributes.xml", {"attributes": attributes}
)
@ -48,7 +48,9 @@ def _get_in_response_to(params):
def _get_subject(params):
"""Insert Subject. Modifies the params dict."""
params["SUBJECT_STATEMENT"] = render_to_string("saml/xml/subject.xml", params)
params["SUBJECT_STATEMENT"] = render_to_string(
"providers/saml/xml/subject.xml", params
)
def get_assertion_xml(template, parameters, signed=False):
@ -80,7 +82,7 @@ def get_response_xml(parameters, saml_provider: SAMLProvider, assertion_id=""):
params["RESPONSE_SIGNATURE"] = ""
_get_in_response_to(params)
raw_response = render_to_string("saml/xml/response.xml", params)
raw_response = render_to_string("providers/saml/xml/response.xml", params)
if not saml_provider.signing_kp:
return raw_response

2
passbook/providers/saml/utils/xml_signing.py
View File

@ -35,4 +35,4 @@ def sign_with_signxml(data: str, provider: "SAMLProvider", reference_uri=None) -
def get_signature_xml() -> str:
"""Returns XML Signature for subject."""
return render_to_string("saml/xml/signature.xml", {})
return render_to_string("providers/saml/xml/signature.xml", {})

31
passbook/providers/saml/views.py
View File

@ -32,6 +32,7 @@ from passbook.policies.engine import PolicyEngine
from passbook.providers.saml.exceptions import CannotHandleAssertion
from passbook.providers.saml.models import SAMLBindings, SAMLProvider
from passbook.providers.saml.processors.types import SAMLResponseParams
from passbook.stages.consent.stage import PLAN_CONTEXT_CONSENT_TEMPLATE
LOGGER = get_logger()
URL_VALIDATOR = URLValidator(schemes=("http", "https"))
@ -87,9 +88,13 @@ class SAMLSSOView(LoginRequiredMixin, SAMLAccessMixin, View):
planner.allow_empty_flows = True
plan = planner.plan(
self.request,
{PLAN_CONTEXT_SSO: True, PLAN_CONTEXT_APPLICATION: self.application},
{
PLAN_CONTEXT_SSO: True,
PLAN_CONTEXT_APPLICATION: self.application,
PLAN_CONTEXT_CONSENT_TEMPLATE: "providers/saml/consent.html",
},
)
plan.stages.append(in_memory_stage(SAMLFlowFinalView))
plan.append(in_memory_stage(SAMLFlowFinalView))
self.request.session[SESSION_KEY_PLAN] = plan
return redirect_with_qs(
"passbook_flows:flow-executor-shell",
@ -188,7 +193,9 @@ class SAMLFlowFinalView(StageView):
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
application: Application = self.executor.plan.context[PLAN_CONTEXT_APPLICATION]
provider: SAMLProvider = application.provider
provider: SAMLProvider = get_object_or_404(
SAMLProvider, pk=application.provider_id
)
# Log Application Authorization
Event.new(
EventAction.AUTHORIZE_APPLICATION,
@ -205,7 +212,7 @@ class SAMLFlowFinalView(StageView):
if provider.sp_binding == SAMLBindings.POST:
return render(
self.request,
"saml/idp/autosubmit_form.html",
"providers/saml/autosubmit_form.html",
{
"url": response.acs_url,
"application": application,
@ -227,7 +234,7 @@ class SAMLFlowFinalView(StageView):
return bad_request_message(request, "Invalid sp_binding specified")
class DescriptorDownloadView(LoginRequiredMixin, SAMLAccessMixin, View):
class DescriptorDownloadView(View):
"""Replies with the XML Metadata IDSSODescriptor."""
@staticmethod
@ -257,18 +264,16 @@ class DescriptorDownloadView(LoginRequiredMixin, SAMLAccessMixin, View):
ctx["cert_public_key"] = strip_pem_header(
provider.signing_kp.certificate_data.replace("\r", "")
).replace("\n", "")
return render_to_string("saml/xml/metadata.xml", ctx)
return render_to_string("providers/saml/xml/metadata.xml", ctx)
def get(self, request: HttpRequest, application_slug: str) -> HttpResponse:
"""Replies with the XML Metadata IDSSODescriptor."""
self.application = get_object_or_404(Application, slug=application_slug)
self.provider: SAMLProvider = get_object_or_404(
SAMLProvider, pk=self.application.provider_id
application = get_object_or_404(Application, slug=application_slug)
provider: SAMLProvider = get_object_or_404(
SAMLProvider, pk=application.provider_id
)
if not self._has_access():
raise PermissionDenied()
try:
metadata = DescriptorDownloadView.get_metadata(request, self.provider)
metadata = DescriptorDownloadView.get_metadata(request, provider)
except Provider.application.RelatedObjectDoesNotExist: # pylint: disable=no-member
return bad_request_message(
request, "Provider is not assigned to an application."
@ -277,5 +282,5 @@ class DescriptorDownloadView(LoginRequiredMixin, SAMLAccessMixin, View):
response = HttpResponse(metadata, content_type="application/xml")
response[
"Content-Disposition"
] = f'attachment; filename="{self.provider.name}_passbook_meta.xml"'
] = f'attachment; filename="{provider.name}_passbook_meta.xml"'
return response

23
passbook/root/settings.py
View File

@ -330,13 +330,24 @@ LOGGING = {
},
"loggers": {},
}
TEST = False
TEST_RUNNER = "xmlrunner.extra.djangotestrunner.XMLTestRunner"
LOG_LEVEL = CONFIG.y("log_level").upper()
TEST_OUTPUT_FILE_NAME = "unittest.xml"
if len(sys.argv) >= 2 and sys.argv[1] == "test":
LOG_LEVEL = "DEBUG"
TEST = True
CELERY_TASK_ALWAYS_EAGER = True
_LOGGING_HANDLER_MAP = {
"": LOG_LEVEL,
"passbook": LOG_LEVEL,
"django": "WARNING",
"celery": "WARNING",
"selenium": "WARNING",
"grpc": LOG_LEVEL,
"oauthlib": LOG_LEVEL,
"oauth2_provider": LOG_LEVEL,
@ -350,18 +361,6 @@ for handler_name, level in _LOGGING_HANDLER_MAP.items():
"propagate": False,
}
TEST = False
TEST_RUNNER = "xmlrunner.extra.djangotestrunner.XMLTestRunner"
TEST_OUTPUT_VERBOSE = 2
TEST_OUTPUT_FILE_NAME = "unittest.xml"
if any("test" in arg for arg in sys.argv):
LOGGER.warning("Testing mode enabled, no logging from now on...")
LOGGING = None
TEST = True
CELERY_TASK_ALWAYS_EAGER = True
_DISALLOWED_ITEMS = [
"INSTALLED_APPS",

6
passbook/sources/oauth/forms.py
View File

@ -84,9 +84,9 @@ class FacebookOAuthSourceForm(OAuthSourceForm):
overrides = {
"provider_type": "facebook",
"request_token_url": "",
"authorization_url": "https://www.facebook.com/v2.8/dialog/oauth",
"access_token_url": "https://graph.facebook.com/v2.8/oauth/access_token",
"profile_url": "https://graph.facebook.com/v2.8/me?fields=name,email,short_name",
"authorization_url": "https://www.facebook.com/v7.0/dialog/oauth",
"access_token_url": "https://graph.facebook.com/v7.0/oauth/access_token",
"profile_url": "https://graph.facebook.com/v7.0/me?fields=id,name,email",
}

15
passbook/sources/oauth/types/facebook.py
View File

@ -1,4 +1,9 @@
"""Facebook OAuth Views"""
from typing import Any, Dict, Optional
from facebook import GraphAPI
from passbook.sources.oauth.clients import OAuth2Client
from passbook.sources.oauth.types.manager import MANAGER, RequestKind
from passbook.sources.oauth.utils import user_get_or_create
from passbook.sources.oauth.views.core import OAuthCallback, OAuthRedirect
@ -14,10 +19,20 @@ class FacebookOAuthRedirect(OAuthRedirect):
}
class FacebookOAuth2Client(OAuth2Client):
"""Facebook OAuth2 Client"""
def get_profile_info(self, token: Dict[str, str]) -> Optional[Dict[str, Any]]:
api = GraphAPI(access_token=token["access_token"])
return api.get_object("me", fields="id,name,email")
@MANAGER.source(kind=RequestKind.callback, name="Facebook")
class FacebookOAuth2Callback(OAuthCallback):
"""Facebook OAuth2 Callback"""
client_class = FacebookOAuth2Client
def get_or_create_user(self, source, access, info):
user_data = {
"username": info.get("name"),

11
passbook/sources/saml/api.py
View File

@ -2,6 +2,7 @@
from rest_framework.serializers import ModelSerializer
from rest_framework.viewsets import ModelViewSet
from passbook.admin.forms.source import SOURCE_FORM_FIELDS
from passbook.sources.saml.models import SAMLSource
@ -11,12 +12,12 @@ class SAMLSourceSerializer(ModelSerializer):
class Meta:
model = SAMLSource
fields = [
"pk",
fields = SOURCE_FORM_FIELDS + [
"issuer",
"idp_url",
"idp_logout_url",
"auto_logout",
"sso_url",
"binding_type",
"slo_url",
"temporary_user_delete_after",
"signing_kp",
]

5
passbook/sources/saml/apps.py
View File

@ -1,5 +1,7 @@
"""Passbook SAML app config"""
from importlib import import_module
from django.apps import AppConfig
@ -10,3 +12,6 @@ class PassbookSourceSAMLConfig(AppConfig):
label = "passbook_sources_saml"
verbose_name = "passbook Sources.SAML"
mountpoint = "source/saml/"
def ready(self):
import_module("passbook.sources.saml.signals")

16
passbook/sources/saml/forms.py
View File

@ -1,8 +1,6 @@
"""passbook SAML SP Forms"""
from django import forms
from django.contrib.admin.widgets import FilteredSelectMultiple
from django.utils.translation import gettext as _
from passbook.admin.forms.source import SOURCE_FORM_FIELDS
from passbook.sources.saml.models import SAMLSource
@ -16,16 +14,16 @@ class SAMLSourceForm(forms.ModelForm):
model = SAMLSource
fields = SOURCE_FORM_FIELDS + [
"issuer",
"idp_url",
"idp_logout_url",
"auto_logout",
"sso_url",
"binding_type",
"slo_url",
"temporary_user_delete_after",
"signing_kp",
]
widgets = {
"name": forms.TextInput(),
"policies": FilteredSelectMultiple(_("policies"), False),
"issuer": forms.TextInput(),
"idp_url": forms.TextInput(),
"idp_logout_url": forms.TextInput(),
"sso_url": forms.TextInput(),
"slo_url": forms.TextInput(),
"temporary_user_delete_after": forms.TextInput(),
}
labels = {"signing_kp": _("Singing Keypair")}

65
passbook/sources/saml/migrations/0003_auto_20200624_1957.py Normal file
View File

@ -0,0 +1,65 @@
# Generated by Django 3.0.7 on 2020-06-24 19:57
import django.db.models.deletion
from django.db import migrations, models
import passbook.providers.saml.utils.time
class Migration(migrations.Migration):
dependencies = [
("passbook_crypto", "0002_create_self_signed_kp"),
("passbook_sources_saml", "0002_auto_20200523_2329"),
]
operations = [
migrations.RemoveField(model_name="samlsource", name="auto_logout",),
migrations.RenameField(
model_name="samlsource", old_name="idp_url", new_name="sso_url",
),
migrations.RenameField(
model_name="samlsource", old_name="idp_logout_url", new_name="slo_url",
),
migrations.AddField(
model_name="samlsource",
name="temporary_user_delete_after",
field=models.TextField(
default="days=1",
help_text="Time offset when temporary users should be deleted. This only applies if your IDP uses the NameID Format 'transient', and the user doesn't log out manually. (Format: hours=1;minutes=2;seconds=3).",
validators=[
passbook.providers.saml.utils.time.timedelta_string_validator
],
verbose_name="Delete temporary users after",
),
),
migrations.AlterField(
model_name="samlsource",
name="signing_kp",
field=models.ForeignKey(
help_text="Certificate Key Pair of the IdP which Assertion's Signature is validated against.",
on_delete=django.db.models.deletion.PROTECT,
to="passbook_crypto.CertificateKeyPair",
verbose_name="Singing Keypair",
),
),
migrations.AlterField(
model_name="samlsource",
name="slo_url",
field=models.URLField(
blank=True,
default=None,
help_text="Optional URL if your IDP supports Single-Logout.",
null=True,
verbose_name="SLO URL",
),
),
migrations.AlterField(
model_name="samlsource",
name="sso_url",
field=models.URLField(
help_text="URL that the initial Login request is sent to.",
verbose_name="SSO URL",
),
),
]

38
passbook/sources/saml/models.py
View File

@ -6,6 +6,7 @@ from django.utils.translation import gettext_lazy as _
from passbook.core.models import Source
from passbook.core.types import UILoginButton
from passbook.crypto.models import CertificateKeyPair
from passbook.providers.saml.utils.time import timedelta_string_validator
class SAMLBindingTypes(models.TextChoices):
@ -25,11 +26,9 @@ class SAMLSource(Source):
help_text=_("Also known as Entity ID. Defaults the Metadata URL."),
)
idp_url = models.URLField(
verbose_name=_("IDP URL"),
help_text=_(
"URL that the initial SAML Request is sent to. Also known as a Binding."
),
sso_url = models.URLField(
verbose_name=_("SSO URL"),
help_text=_("URL that the initial Login request is sent to."),
)
binding_type = models.CharField(
max_length=100,
@ -37,19 +36,34 @@ class SAMLSource(Source):
default=SAMLBindingTypes.Redirect,
)
idp_logout_url = models.URLField(
default=None, blank=True, null=True, verbose_name=_("IDP Logout URL")
slo_url = models.URLField(
default=None,
blank=True,
null=True,
verbose_name=_("SLO URL"),
help_text=_("Optional URL if your IDP supports Single-Logout."),
)
temporary_user_delete_after = models.TextField(
default="days=1",
verbose_name=_("Delete temporary users after"),
validators=[timedelta_string_validator],
help_text=_(
(
"Time offset when temporary users should be deleted. This only applies if your IDP "
"uses the NameID Format 'transient', and the user doesn't log out manually. "
"(Format: hours=1;minutes=2;seconds=3)."
)
),
)
auto_logout = models.BooleanField(default=False)
signing_kp = models.ForeignKey(
CertificateKeyPair,
default=None,
null=True,
verbose_name=_("Singing Keypair"),
help_text=_(
"Certificate Key Pair of the IdP which Assertions are validated against."
"Certificate Key Pair of the IdP which Assertion's Signature is validated against."
),
on_delete=models.SET_NULL,
on_delete=models.PROTECT,
)
form = "passbook.sources.saml.forms.SAMLSourceForm"

107
passbook/sources/saml/processors/base.py
View File

@ -1,5 +1,5 @@
"""passbook saml source processor"""
from typing import TYPE_CHECKING, Optional
from typing import TYPE_CHECKING, Dict
from defusedxml import ElementTree
from django.http import HttpRequest, HttpResponse
@ -7,6 +7,7 @@ from signxml import XMLVerifier
from structlog import get_logger
from passbook.core.models import User
from passbook.flows.models import Flow
from passbook.flows.planner import (
PLAN_CONTEXT_PENDING_USER,
PLAN_CONTEXT_SSO,
@ -20,6 +21,13 @@ from passbook.sources.saml.exceptions import (
UnsupportedNameIDFormat,
)
from passbook.sources.saml.models import SAMLSource
from passbook.sources.saml.processors.constants import (
SAML_NAME_ID_FORMAT_EMAIL,
SAML_NAME_ID_FORMAT_PRESISTENT,
SAML_NAME_ID_FORMAT_TRANSIENT,
SAML_NAME_ID_FORMAT_WINDOWS,
SAML_NAME_ID_FORMAT_X509,
)
from passbook.stages.password.stage import PLAN_CONTEXT_AUTHENTICATION_BACKEND
from passbook.stages.prompt.stage import PLAN_CONTEXT_PROMPT
@ -60,53 +68,92 @@ class Processor:
self._root_xml, x509_cert=self._source.signing_kp.certificate_data
)
def _get_email(self) -> Optional[str]:
"""
Returns the email out of the response.
def _handle_name_id_transient(self, request: HttpRequest) -> HttpResponse:
"""Handle a NameID with the Format of Transient. This is a bit more complex than other
formats, as we need to create a temporary User that is used in the session. This
user has an attribute that refers to our Source for cleanup. The user is also deleted
on logout and periodically."""
# Create a temporary User
name_id = self._get_name_id().text
user: User = User.objects.create(
username=name_id,
attributes={
"saml": {"source": self._source.pk.hex, "delete_on_logout": True}
},
)
LOGGER.debug("Created temporary user for NameID Transient", username=name_id)
user.set_unusable_password()
user.save()
return self._flow_response(
request,
self._source.authentication_flow,
**{
PLAN_CONTEXT_PENDING_USER: user,
PLAN_CONTEXT_AUTHENTICATION_BACKEND: DEFAULT_BACKEND,
},
)
At present, response must pass the email address as the Subject, eg.:
<saml:Subject>
<saml:NameID Format="urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"
SPNameQualifier=""
>email@example.com</saml:NameID>
"""
def _get_name_id(self) -> "Element":
"""Get NameID Element"""
assertion = self._root.find("{urn:oasis:names:tc:SAML:2.0:assertion}Assertion")
subject = assertion.find("{urn:oasis:names:tc:SAML:2.0:assertion}Subject")
name_id = subject.find("{urn:oasis:names:tc:SAML:2.0:assertion}NameID")
name_id_format = name_id.attrib["Format"]
if name_id_format != "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress":
if name_id is None:
raise ValueError("NameID Element not found!")
return name_id
def _get_name_id_filter(self) -> Dict[str, str]:
"""Returns the subject's NameID as a Filter for the `User`"""
name_id_el = self._get_name_id()
name_id = name_id_el.text
if not name_id:
raise UnsupportedNameIDFormat("Subject's NameID is empty.")
_format = name_id_el.attrib["Format"]
if _format == SAML_NAME_ID_FORMAT_EMAIL:
return {"email": name_id}
if _format == SAML_NAME_ID_FORMAT_PRESISTENT:
return {"username": name_id}
if _format == SAML_NAME_ID_FORMAT_X509:
# This attribute is statically set by the LDAP source
return {"attributes__distinguishedName": name_id}
if _format == SAML_NAME_ID_FORMAT_WINDOWS:
if "\\" in name_id:
name_id = name_id.split("\\")[1]
return {"username": name_id}
raise UnsupportedNameIDFormat(
f"Assertion contains NameID with unsupported format {name_id_format}."
f"Assertion contains NameID with unsupported format {_format}."
)
return name_id.text
def prepare_flow(self, request: HttpRequest) -> HttpResponse:
"""Prepare flow plan depending on whether or not the user exists"""
email = self._get_email()
matching_users = User.objects.filter(email=email)
name_id = self._get_name_id()
# transient NameIDs are handeled seperately as they don't have to go through flows.
if name_id.attrib["Format"] == SAML_NAME_ID_FORMAT_TRANSIENT:
return self._handle_name_id_transient(request)
name_id_filter = self._get_name_id_filter()
matching_users = User.objects.filter(**name_id_filter)
if matching_users.exists():
# User exists already, switch to authentication flow
flow = self._source.authentication_flow
request.session[SESSION_KEY_PLAN] = FlowPlanner(flow).plan(
return self._flow_response(
request,
{
# Data for authentication
self._source.authentication_flow,
**{
PLAN_CONTEXT_PENDING_USER: matching_users.first(),
PLAN_CONTEXT_AUTHENTICATION_BACKEND: DEFAULT_BACKEND,
PLAN_CONTEXT_SSO: True,
},
)
else:
flow = self._source.enrollment_flow
request.session[SESSION_KEY_PLAN] = FlowPlanner(flow).plan(
return self._flow_response(
request,
{
# Data for enrollment
PLAN_CONTEXT_PROMPT: {"username": email, "email": email},
PLAN_CONTEXT_SSO: True,
},
self._source.enrollment_flow,
**{PLAN_CONTEXT_PROMPT: name_id_filter},
)
def _flow_response(
self, request: HttpRequest, flow: Flow, **kwargs
) -> HttpResponse:
kwargs[PLAN_CONTEXT_SSO] = True
request.session[SESSION_KEY_PLAN] = FlowPlanner(flow).plan(request, kwargs,)
return redirect_with_qs(
"passbook_flows:flow-executor-shell", request.GET, flow_slug=flow.slug,
)

8
passbook/sources/saml/processors/constants.py Normal file
View File

@ -0,0 +1,8 @@
"""SAML Source processor constants"""
SAML_NAME_ID_FORMAT_EMAIL = "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"
SAML_NAME_ID_FORMAT_PRESISTENT = "urn:oasis:names:tc:SAML:2.0:nameid-format:persistent"
SAML_NAME_ID_FORMAT_X509 = "urn:oasis:names:tc:SAML:2.0:nameid-format:X509SubjectName"
SAML_NAME_ID_FORMAT_WINDOWS = (
"urn:oasis:names:tc:SAML:2.0:nameid-format:WindowsDomainQualifiedName"
)
SAML_NAME_ID_FORMAT_TRANSIENT = "urn:oasis:names:tc:SAML:2.0:nameid-format:transient"

9
passbook/sources/saml/settings.py Normal file
View File

@ -0,0 +1,9 @@
"""saml source settings"""
from celery.schedules import crontab
CELERY_BEAT_SCHEDULE = {
"saml_source_cleanup": {
"task": "passbook.sources.saml.tasks.clean_temporary_users",
"schedule": crontab(minute="*/5"),
}
}

22
passbook/sources/saml/signals.py Normal file
View File

@ -0,0 +1,22 @@
"""passbook saml source signal listener"""
from django.contrib.auth.signals import user_logged_out
from django.dispatch import receiver
from django.http import HttpRequest
from structlog import get_logger
from passbook.core.models import User
LOGGER = get_logger()
@receiver(user_logged_out)
# pylint: disable=unused-argument
def on_user_logged_out(sender, request: HttpRequest, user: User, **_):
"""Delete temporary user if the `delete_on_logout` flag is enabled"""
if not user:
return
if "saml" in user.attributes:
if "delete_on_logout" in user.attributes["saml"]:
if user.attributes["saml"]["delete_on_logout"]:
LOGGER.debug("Deleted temporary user", user=user)
user.delete()

32
passbook/sources/saml/tasks.py Normal file
View File

@ -0,0 +1,32 @@
"""passbook saml source tasks"""
from django.utils.timezone import now
from structlog import get_logger
from passbook.core.models import User
from passbook.providers.saml.utils.time import timedelta_from_string
from passbook.root.celery import CELERY_APP
from passbook.sources.saml.models import SAMLSource
LOGGER = get_logger()
@CELERY_APP.task()
def clean_temporary_users():
"""Remove old temporary users"""
_now = now()
for user in User.objects.filter(attributes__saml__isnull=False):
sources = SAMLSource.objects.filter(
pk=user.attributes.get("saml", {}).get("source", "")
)
if not sources.exists():
LOGGER.warning(
"User has an invalid SAML Source and won't be deleted!", user=user
)
source = sources.first()
source_delta = timedelta_from_string(source.temporary_user_delete_after)
if _now - user.last_login >= source_delta:
LOGGER.debug(
"User is expired and will be deleted.", user=user, delta=source_delta
)
# TODO: Check if user is signed in anywhere?
user.delete()

2
passbook/sources/saml/templates/saml/sp/login.html
View File

@ -14,7 +14,7 @@
<form method="POST" action="{{ request_url }}">
{% csrf_token %}
<input type="hidden" name="SAMLRequest" value="{{ request }}" />
<input type="hidden" name="RelayState" value="{{ token }}" />
<input type="hidden" name="RelayState" value="{{ relay_state }}" />
<div class="login-group">
<h3>
{% blocktrans with remote=source.name %}

30
passbook/sources/saml/views.py
View File

@ -1,16 +1,18 @@
"""saml sp views"""
from django.contrib.auth import logout
from django.contrib.auth.mixins import LoginRequiredMixin
from django.http import Http404, HttpRequest, HttpResponse
from django.shortcuts import get_object_or_404, redirect, render
from django.utils.decorators import method_decorator
from django.utils.http import urlencode
from django.views import View
from django.views.decorators.csrf import csrf_exempt
from signxml import InvalidSignature
from signxml.util import strip_pem_header
from passbook.lib.views import bad_request_message
from passbook.providers.saml.utils import get_random_id, render_xml
from passbook.providers.saml.utils.encoding import nice64
from passbook.providers.saml.utils.encoding import deflate_and_base64_encode, nice64
from passbook.providers.saml.utils.time import get_time_string
from passbook.sources.saml.exceptions import (
MissingSAMLResponse,
@ -30,27 +32,29 @@ class InitiateView(View):
source: SAMLSource = get_object_or_404(SAMLSource, slug=source_slug)
if not source.enabled:
raise Http404
sso_destination = request.GET.get("next", None)
request.session["sso_destination"] = sso_destination
relay_state = request.GET.get("next", "")
request.session["sso_destination"] = relay_state
parameters = {
"ACS_URL": build_full_url("acs", request, source),
"DESTINATION": source.idp_url,
"DESTINATION": source.sso_url,
"AUTHN_REQUEST_ID": get_random_id(),
"ISSUE_INSTANT": get_time_string(),
"ISSUER": get_issuer(request, source),
}
authn_req = get_authnrequest_xml(parameters, signed=False)
_request = nice64(str.encode(authn_req))
if source.binding_type == SAMLBindingTypes.Redirect:
return redirect(source.idp_url + "?" + urlencode({"SAMLRequest": _request}))
_request = deflate_and_base64_encode(authn_req.encode())
url_args = urlencode({"SAMLRequest": _request, "RelayState": relay_state})
return redirect(f"{source.sso_url}?{url_args}")
if source.binding_type == SAMLBindingTypes.POST:
_request = nice64(authn_req.encode())
return render(
request,
"saml/sp/login.html",
{
"request_url": source.idp_url,
"request_url": source.sso_url,
"request": _request,
"token": sso_destination,
"relay_state": relay_state,
"source": source,
},
)
@ -71,6 +75,8 @@ class ACSView(View):
processor.parse(request)
except MissingSAMLResponse as exc:
return bad_request_message(request, str(exc))
except InvalidSignature as exc:
return bad_request_message(request, str(exc))
try:
return processor.prepare_flow(request)
@ -78,11 +84,12 @@ class ACSView(View):
return bad_request_message(request, str(exc))
class SLOView(View):
class SLOView(LoginRequiredMixin, View):
"""Single-Logout-View"""
def dispatch(self, request: HttpRequest, source_slug: str) -> HttpResponse:
"""Replies with an XHTML SSO Request."""
# TODO: Replace with flows
source: SAMLSource = get_object_or_404(SAMLSource, slug=source_slug)
if not source.enabled:
raise Http404
@ -90,10 +97,7 @@ class SLOView(View):
return render(
request,
"saml/sp/sso_single_logout.html",
{
"idp_logout_url": source.idp_logout_url,
"autosubmit": source.auto_logout,
},
{"idp_logout_url": source.slo_url},
)

17
passbook/stages/consent/stage.py
View File

@ -1,25 +1,30 @@
"""passbook consent stage"""
from typing import Any, Dict
from typing import Any, Dict, List
from django.views.generic import FormView
from passbook.flows.stage import StageView
from passbook.lib.utils.template import render_to_string
from passbook.stages.consent.forms import ConsentForm
PLAN_CONTEXT_CONSENT_TEMPLATE = "consent_template"
class ConsentStage(FormView, StageView):
"""Simple consent checker."""
body_template_name: str
form_class = ConsentForm
def get_context_data(self, **kwargs: Dict[str, Any]) -> Dict[str, Any]:
kwargs = super().get_context_data(**kwargs)
if self.body_template_name:
kwargs["body"] = render_to_string(self.body_template_name, kwargs)
kwargs["current_stage"] = self.executor.current_stage
kwargs["context"] = self.executor.plan.context
return kwargs
def get_template_names(self) -> List[str]:
if PLAN_CONTEXT_CONSENT_TEMPLATE in self.executor.plan.context:
template_name = self.executor.plan.context[PLAN_CONTEXT_CONSENT_TEMPLATE]
return [template_name]
return super().get_template_names()
def form_valid(self, form):
return self.executor.stage_ok()

2
passbook/stages/email/models.py
View File

@ -21,7 +21,7 @@ class EmailTemplates(models.TextChoices):
class EmailStage(Stage):
"""email stage"""
"""Email-based verification."""
host = models.TextField(default="localhost")
port = models.IntegerField(default=25)

55
passbook/stages/email/stage.py
View File

@ -13,12 +13,15 @@ from structlog import get_logger
from passbook.core.models import Token
from passbook.flows.planner import PLAN_CONTEXT_PENDING_USER
from passbook.flows.stage import StageView
from passbook.flows.views import SESSION_KEY_GET
from passbook.stages.email.forms import EmailStageSendForm
from passbook.stages.email.models import EmailStage
from passbook.stages.email.tasks import send_mails
from passbook.stages.email.utils import TemplateEmailMessage
LOGGER = get_logger()
QS_KEY_TOKEN = "token"
PLAN_CONTEXT_EMAIL_SENT = "email_sent"
class EmailStageView(FormView, StageView):
@ -30,34 +33,25 @@ class EmailStageView(FormView, StageView):
def get_full_url(self, **kwargs) -> str:
"""Get full URL to be used in template"""
base_url = reverse(
"passbook_flows:flow-executor",
"passbook_flows:flow-executor-shell",
kwargs={"flow_slug": self.executor.flow.slug},
)
relative_url = f"{base_url}?{urlencode(kwargs)}"
return self.request.build_absolute_uri(relative_url)
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
if QS_KEY_TOKEN in request.GET:
token = get_object_or_404(Token, pk=request.GET[QS_KEY_TOKEN])
self.executor.plan.context[PLAN_CONTEXT_PENDING_USER] = token.user
token.delete()
messages.success(request, _("Successfully verified Email."))
return self.executor.stage_ok()
return super().get(request, *args, **kwargs)
def form_invalid(self, form: EmailStageSendForm) -> HttpResponse:
if PLAN_CONTEXT_PENDING_USER not in self.executor.plan.context:
messages.error(self.request, _("No pending user."))
return super().form_invalid(form)
def send_email(self):
"""Helper function that sends the actual email. Implies that you've
already checked that there is a pending user."""
pending_user = self.executor.plan.context[PLAN_CONTEXT_PENDING_USER]
current_stage: EmailStage = self.executor.current_stage
valid_delta = timedelta(
minutes=self.executor.current_stage.token_expiry + 1
minutes=current_stage.token_expiry + 1
) # + 1 because django timesince always rounds down
token = Token.objects.create(user=pending_user, expires=now() + valid_delta)
# Send mail to user
message = TemplateEmailMessage(
subject=_("passbook - Password Recovery"),
template_name=self.executor.current_stage.template,
subject=_(current_stage.subject),
template_name=current_stage.template,
to=[pending_user.email],
template_context={
"url": self.get_full_url(**{QS_KEY_TOKEN: token.pk.hex}),
@ -65,7 +59,32 @@ class EmailStageView(FormView, StageView):
"expires": token.expires,
},
)
send_mails(self.executor.current_stage, message)
send_mails(current_stage, message)
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
# Check if the user came back from the email link to verify
if QS_KEY_TOKEN in request.session.get(SESSION_KEY_GET, {}):
token = get_object_or_404(
Token, pk=request.session[SESSION_KEY_GET][QS_KEY_TOKEN]
)
self.executor.plan.context[PLAN_CONTEXT_PENDING_USER] = token.user
token.delete()
messages.success(request, _("Successfully verified Email."))
return self.executor.stage_ok()
if PLAN_CONTEXT_PENDING_USER not in self.executor.plan.context:
messages.error(self.request, _("No pending user."))
return self.executor.stage_invalid()
# Check if we've already sent the initial e-mail
if PLAN_CONTEXT_EMAIL_SENT not in self.executor.plan.context:
self.send_email()
self.executor.plan.context[PLAN_CONTEXT_EMAIL_SENT] = True
return super().get(request, *args, **kwargs)
def form_invalid(self, form: EmailStageSendForm) -> HttpResponse:
if PLAN_CONTEXT_PENDING_USER not in self.executor.plan.context:
messages.error(self.request, _("No pending user."))
return super().form_invalid(form)
self.send_email()
# We can't call stage_ok yet, as we're still waiting
# for the user to click the link in the email
return super().form_invalid(form)

6
passbook/stages/email/templates/stages/email/for_email/account_confirm.html → passbook/stages/email/templates/stages/email/for_email/account_confirmation.html
View File

@ -1,6 +1,6 @@
{% extends 'email/base.html' %}
{% extends 'stages/email/for_email/base.html' %}
{% load inline %}
{% load passbook_stages_email %}
{% load i18n %}
{% block content %}
@ -18,7 +18,7 @@
<table role="presentation" border="0" cellpadding="0" cellspacing="0">
<tbody>
<tr>
<td> <a href="{{ url }}" rel="noopener noreferrer" target="_blank">{% trans 'Confirm Account' %}</a> </td>
<td> <a id="confirm" href="{{ url }}" rel="noopener noreferrer" target="_blank">{% trans 'Confirm Account' %}</a> </td>
</tr>
</tbody>
</table>

2
passbook/stages/email/templates/stages/email/for_email/generic_email.html
View File

@ -1,4 +1,4 @@
{% extends "email/base.html" %}
{% extends "stages/email/for_email/base.html" %}
{% block content %}
<tr>

2
passbook/stages/email/templates/stages/email/for_email/password_reset.html
View File

@ -23,7 +23,7 @@
<table role="presentation" border="0" cellpadding="0" cellspacing="0">
<tbody>
<tr>
<td> <a href="{{ url }}" rel="noopener noreferrer" target="_blank">{% trans 'Reset Password' %}</a> </td>
<td> <a id="confirm" href="{{ url }}" rel="noopener noreferrer" target="_blank">{% trans 'Reset Password' %}</a> </td>
</tr>
</tbody>
</table>

2
passbook/stages/email/templates/stages/email/waiting_message.html
View File

@ -15,7 +15,7 @@
{% block beneath_form %}
{% endblock %}
<div class="pf-c-form__group pf-m-action">
<button class="pf-c-button pf-m-primary pf-m-block" type="submit">{% trans "Send Recovery Email." %}</button>
<button class="pf-c-button pf-m-block" type="submit">{% trans "Send Email again." %}</button>
</div>
</form>
{% endblock %}

14
passbook/stages/email/tests.py
View File

@ -83,7 +83,7 @@ class TestEmailStage(TestCase):
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(mail.outbox), 1)
self.assertEqual(mail.outbox[0].subject, "passbook - Password Recovery")
self.assertEqual(mail.outbox[0].subject, "passbook")
def test_token(self):
"""Test with token"""
@ -97,12 +97,20 @@ class TestEmailStage(TestCase):
session.save()
with patch("passbook.flows.views.FlowExecutorView.cancel", MagicMock()):
# Call the executor shell to preseed the session
url = reverse(
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
"passbook_flows:flow-executor-shell",
kwargs={"flow_slug": self.flow.slug},
)
token = Token.objects.get(user=self.user)
url += f"?{QS_KEY_TOKEN}={token.pk.hex}"
response = self.client.get(url)
self.client.get(url)
# Call the actual executor to get the JSON Response
response = self.client.get(
reverse(
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
)
)
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(

2
passbook/stages/identification/stage.py
View File

@ -49,7 +49,7 @@ class IdentificationStageView(FormView, StageView):
# Check all enabled source, add them if they have a UI Login button.
kwargs["sources"] = []
sources = (
sources: List[Source] = (
Source.objects.filter(enabled=True).order_by("name").select_subclasses()
)
for source in sources:

4
passbook/stages/identification/templates/stages/identification/login.html
View File

@ -43,12 +43,12 @@
{% if enroll_url %}
<p class="pf-c-login__main-footer-band-item">
{% trans 'Need an account?' %}
<a href="{{ enroll_url }}">{% trans 'Sign up.' %}</a>
<a role="enroll" href="{{ enroll_url }}">{% trans 'Sign up.' %}</a>
</p>
{% endif %}
{% if recovery_url %}
<p class="pf-c-login__main-footer-band-item">
<a href="{{ recovery_url }}">{% trans 'Forgot username or password?' %}</a>
<a role="recovery" href="{{ recovery_url }}">{% trans 'Forgot username or password?' %}</a>
</p>
{% endif %}
</div>

2
scripts/coverage.sh
View File

@ -1,5 +1,5 @@
#!/bin/bash -xe
coverage run --concurrency=multiprocessing manage.py test --failfast
coverage run --concurrency=multiprocessing manage.py test passbook --failfast
coverage combine
coverage html
coverage report

6
scripts/pre-commit.sh
View File

@ -1,9 +1,9 @@
#!/bin/bash -xe
isort -rc passbook
isort -rc .
pyright
black passbook
black .
./manage.py generate_swagger -o swagger.yaml -f yaml
scripts/coverage.sh
bandit -r passbook
bandit -r .
pylint passbook
prospector

200
swagger.yaml
View File

@ -2951,6 +2951,133 @@ paths:
required: true
type: string
format: uuid
/sources/saml/:
get:
operationId: sources_saml_list
description: SAMLSource Viewset
parameters:
- name: ordering
in: query
description: Which field to use when ordering the results.
required: false
type: string
- name: search
in: query
description: A search term.
required: false
type: string
- name: limit
in: query
description: Number of results to return per page.
required: false
type: integer
- name: offset
in: query
description: The initial index from which to return the results.
required: false
type: integer
responses:
'200':
description: ''
schema:
required:
- count
- results
type: object
properties:
count:
type: integer
next:
type: string
format: uri
x-nullable: true
previous:
type: string
format: uri
x-nullable: true
results:
type: array
items:
$ref: '#/definitions/SAMLSource'
tags:
- sources
post:
operationId: sources_saml_create
description: SAMLSource Viewset
parameters:
- name: data
in: body
required: true
schema:
$ref: '#/definitions/SAMLSource'
responses:
'201':
description: ''
schema:
$ref: '#/definitions/SAMLSource'
tags:
- sources
parameters: []
/sources/saml/{pbm_uuid}/:
get:
operationId: sources_saml_read
description: SAMLSource Viewset
parameters: []
responses:
'200':
description: ''
schema:
$ref: '#/definitions/SAMLSource'
tags:
- sources
put:
operationId: sources_saml_update
description: SAMLSource Viewset
parameters:
- name: data
in: body
required: true
schema:
$ref: '#/definitions/SAMLSource'
responses:
'200':
description: ''
schema:
$ref: '#/definitions/SAMLSource'
tags:
- sources
patch:
operationId: sources_saml_partial_update
description: SAMLSource Viewset
parameters:
- name: data
in: body
required: true
schema:
$ref: '#/definitions/SAMLSource'
responses:
'200':
description: ''
schema:
$ref: '#/definitions/SAMLSource'
tags:
- sources
delete:
operationId: sources_saml_delete
description: SAMLSource Viewset
parameters: []
responses:
'204':
description: ''
tags:
- sources
parameters:
- name: pbm_uuid
in: path
description: A UUID string identifying this SAML Source.
required: true
type: string
format: uuid
/stages/all/:
get:
operationId: stages_all_list
@ -5804,6 +5931,79 @@ definitions:
title: Consumer secret
type: string
minLength: 1
SAMLSource:
required:
- name
- slug
- sso_url
- signing_kp
type: object
properties:
name:
title: Name
description: Source's display Name.
type: string
minLength: 1
slug:
title: Slug
description: Internal source name, used in URLs.
type: string
format: slug
pattern: ^[-a-zA-Z0-9_]+$
maxLength: 50
minLength: 1
enabled:
title: Enabled
type: boolean
authentication_flow:
title: Authentication flow
description: Flow to use when authenticating existing users.
type: string
format: uuid
x-nullable: true
enrollment_flow:
title: Enrollment flow
description: Flow to use when enrolling new users.
type: string
format: uuid
x-nullable: true
issuer:
title: Issuer
description: Also known as Entity ID. Defaults the Metadata URL.
type: string
sso_url:
title: SSO URL
description: URL that the initial Login request is sent to.
type: string
format: uri
maxLength: 200
minLength: 1
binding_type:
title: Binding type
type: string
enum:
- REDIRECT
- POST
slo_url:
title: SLO URL
description: Optional URL if your IDP supports Single-Logout.
type: string
format: uri
maxLength: 200
x-nullable: true
temporary_user_delete_after:
title: Delete temporary users after
description: "Time offset when temporary users should be deleted. This only\
\ applies if your IDP uses the NameID Format 'transient', and the user doesn't\
\ log out manually. (Format: hours=1;minutes=2;seconds=3)."
type: string
minLength: 1
signing_kp:
title: Singing Keypair
description: Certificate Key Pair of the IdP which Assertion's Signature is
validated against.
type: string
format: uuid
Stage:
required:
- name