4594239788

Build Type

github

Committed by GitHub

Commit Message

Merge 93719774d into 75538c8bc

Pull Request Pull Request #772: bencher

Run Details

134 of 134 new or added lines in 1 file covered. (100.0%)

96125 of 107670 relevant lines covered (89.28%)

72821.51 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.06

/services/src/pro/users/postgres_userdb.rs

use crate::contexts::SessionId;
use crate::error::Result;
use crate::pro::contexts::PostgresDb;
use crate::pro::permissions::{Role, RoleId};
use crate::pro::users::oidc::ExternalUserClaims;
use crate::pro::users::{
    User, UserCredentials, UserDb, UserId, UserInfo, UserRegistration, UserSession,
};
use crate::projects::{ProjectId, STRectangle};
use crate::util::Identifier;
use crate::{error, pro::contexts::PostgresContext};
use async_trait::async_trait;

use bb8_postgres::{
    tokio_postgres::tls::MakeTlsConnect, tokio_postgres::tls::TlsConnect, tokio_postgres::Socket,
};
use geoengine_datatypes::primitives::Duration;
use pwhash::bcrypt;
use snafu::ensure;
use uuid::Uuid;

use super::userdb::{RoleDb, UserAuth};

#[async_trait]
impl<Tls> UserAuth for PostgresContext<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    // TODO: clean up expired sessions?

    async fn register_user(&self, user: UserRegistration) -> Result<UserId> {
        let mut conn = self.pool.get().await?;

        let tx = conn.build_transaction().start().await?;

        let user = User::from(user);

        let stmt = tx
            .prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user.id, &user.email]).await?;

        let stmt = tx
            .prepare(
                "INSERT INTO users (id, email, password_hash, real_name, quota_available, active) VALUES ($1, $2, $3, $4, $5, $6);",
            )
            .await?;

        let quota_available =
            crate::util::config::get_config_element::<crate::pro::util::config::User>()?
                .default_available_quota;

        tx.execute(
            &stmt,
            &[
                &user.id,
                &user.email,
                &user.password_hash,
                &user.real_name,
                &quota_available,
                &user.active,
            ],
        )
        .await?;

        let stmt = tx
            .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user.id, &user.id]).await?;

        let stmt = tx
            .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user.id, &Role::registered_user_role_id()])
            .await?;

        tx.commit().await?;

        Ok(user.id)
    }

    async fn create_anonymous_session(&self) -> Result<UserSession> {
        let mut conn = self.pool.get().await?;

        let tx = conn.build_transaction().start().await?;

        let user_id = UserId::new();

        let stmt = tx
            .prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user_id, &format!("anonymous_user_{user_id}")])
            .await?;

        let quota_available =
            crate::util::config::get_config_element::<crate::pro::util::config::User>()?
                .default_available_quota;

        let stmt = tx
            .prepare("INSERT INTO users (id, quota_available, active) VALUES ($1, $2, TRUE);")
            .await?;

        tx.execute(&stmt, &[&user_id, &quota_available]).await?;

        let stmt = tx
            .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user_id, &user_id]).await?;

        let stmt = tx
            .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
            .await?;
        tx.execute(&stmt, &[&user_id, &Role::anonymous_role_id()])
            .await?;

        let session_id = SessionId::new();
        let stmt = tx
            .prepare(
                "
                INSERT INTO sessions (id, user_id, created, valid_until)
                VALUES ($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3)) 
                RETURNING created, valid_until;",
            )
            .await?;

        // TODO: load from config
        let session_duration = chrono::Duration::days(30);
        let row = tx
            .query_one(
                &stmt,
                &[
                    &session_id,
                    &user_id,
                    &(session_duration.num_seconds() as f64),
                ],
            )
            .await?;

        tx.commit().await?;

        Ok(UserSession {
            id: session_id,
            user: UserInfo {
                id: user_id,
                email: None,
                real_name: None,
            },
            created: row.get(0),
            valid_until: row.get(1),
            project: None,
            view: None,
            roles: vec![user_id.into(), Role::anonymous_role_id()],
        })
    }

    async fn login(&self, user_credentials: UserCredentials) -> Result<UserSession> {
        let conn = self.pool.get().await?;
        let stmt = conn
            .prepare("SELECT id, password_hash, email, real_name FROM users WHERE email = $1;")
            .await?;

        let row = conn
            .query_one(&stmt, &[&user_credentials.email])
            .await
            .map_err(|_error| error::Error::LoginFailed)?;

        let user_id = UserId(row.get(0));
        let password_hash = row.get(1);
        let email = row.get(2);
        let real_name = row.get(3);

        if bcrypt::verify(user_credentials.password, password_hash) {
            let session_id = SessionId::new();
            let stmt = conn
                .prepare(
                    "
                INSERT INTO sessions (id, user_id, created, valid_until)
                VALUES ($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3)) 
                RETURNING created, valid_until;",
                )
                .await?;

            // TODO: load from config
            let session_duration = chrono::Duration::days(30);
            let row = conn
                .query_one(
                    &stmt,
                    &[
                        &session_id,
                        &user_id,
                        &(session_duration.num_seconds() as f64),
                    ],
                )
                .await?;

            let stmt = conn
                .prepare("SELECT role_id FROM user_roles WHERE user_id = $1;")
                .await?;

            let rows = conn
                .query(&stmt, &[&user_id])
                .await
                .map_err(|_error| error::Error::LoginFailed)?;

            let roles = rows.into_iter().map(|row| row.get(0)).collect();

            Ok(UserSession {
                id: session_id,
                user: UserInfo {
                    id: user_id,
                    email,
                    real_name,
                },
                created: row.get(0),
                valid_until: row.get(1),
                project: None,
                view: None,
                roles,
            })
        } else {
            Err(error::Error::LoginFailed)
        }
    }

    async fn login_external(
        &self,
        user: ExternalUserClaims,
        duration: Duration,
    ) -> Result<UserSession> {
        let mut conn = self.pool.get().await?;
        let stmt = conn
            .prepare("SELECT id, external_id, email, real_name FROM external_users WHERE external_id = $1;")
            .await?;

        let row = conn
            .query_opt(&stmt, &[&user.external_id.to_string()])
            .await
            .map_err(|_error| error::Error::LoginFailed)?;

        let user_id = match row {
            Some(row) => UserId(row.get(0)),
            None => {
                let tx = conn.build_transaction().start().await?;

                let user_id = UserId::new();

                let stmt = tx
                    .prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")
                    .await?;
                tx.execute(&stmt, &[&user_id, &user.email]).await?;

                let quota_available =
                    crate::util::config::get_config_element::<crate::pro::util::config::User>()?
                        .default_available_quota;

                //TODO: Inconsistent to hashmap implementation, where an external user is not part of the user database.
                //TODO: A user might be able to login without external login using this (internal) id. Would be a problem with anonymous users as well.
                let stmt = tx
                    .prepare(
                        "INSERT INTO users (id, quota_available, active) VALUES ($1, $2, TRUE);",
                    )
                    .await?;
                tx.execute(&stmt, &[&user_id, &quota_available]).await?;

                let stmt = tx
                    .prepare(
                        "INSERT INTO external_users (id, external_id, email, real_name, active) VALUES ($1, $2, $3, $4, $5);",
                    )
                    .await?;

                tx.execute(
                    &stmt,
                    &[
                        &user_id,
                        &user.external_id.to_string(),
                        &user.email,
                        &user.real_name,
                        &true,
                    ],
                )
                .await?;

                let stmt = tx
                    .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
                    .await?;
                tx.execute(&stmt, &[&user_id, &user_id]).await?;

                let stmt = tx
                    .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
                    .await?;
                tx.execute(&stmt, &[&user_id, &Role::registered_user_role_id()])
                    .await?;

                tx.commit().await?;

                user_id
            }
        };

        let session_id = SessionId::new();
        let stmt = conn
            .prepare(
                "
            INSERT INTO sessions (id, user_id, created, valid_until)
            VALUES ($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3))
            RETURNING created, valid_until;",
            )
            .await?; //TODO: Check documentation if inconsistent to hashmap implementation - would happen if CURRENT_TIMESTAMP is called twice in postgres for a single query. Worked in tests.

        let row = conn
            .query_one(
                &stmt,
                &[&session_id, &user_id, &(duration.num_seconds() as f64)],
            )
            .await?;

        let stmt = conn
            .prepare("SELECT role_id FROM user_roles WHERE user_id = $1;")
            .await?;

        let rows = conn
            .query(&stmt, &[&user_id])
            .await
            .map_err(|_error| error::Error::LoginFailed)?;

        let roles = rows.into_iter().map(|row| row.get(0)).collect();

        Ok(UserSession {
            id: session_id,
            user: UserInfo {
                id: user_id,
                email: Some(user.email.clone()),
                real_name: Some(user.real_name.clone()),
            },
            created: row.get(0),
            valid_until: row.get(1),
            project: None,
            view: None,
            roles,
        })
    }

    async fn user_session_by_id(&self, session: SessionId) -> Result<UserSession> {
        let mut conn = self.pool.get().await?;

        let tx = conn.build_transaction().start().await?;

        let stmt = tx
            .prepare(
                "
            SELECT 
                u.id,   
                u.email,
                u.real_name,             
                s.created, 
                s.valid_until, 
                s.project_id,
                s.view
            FROM sessions s JOIN users u ON (s.user_id = u.id)
            WHERE s.id = $1 AND CURRENT_TIMESTAMP < s.valid_until;",
            )
            .await?;

        let row = tx
            .query_one(&stmt, &[&session])
            .await
            .map_err(|_error| error::Error::InvalidSession)?;

        let mut session = UserSession {
            id: session,
            user: UserInfo {
                id: row.get(0),
                email: row.get(1),
                real_name: row.get(2),
            },
            created: row.get(3),
            valid_until: row.get(4),
            project: row.get::<usize, Option<Uuid>>(5).map(ProjectId),
            view: row.get(6),
            roles: vec![],
        };

        let stmt = tx
            .prepare(
                "
            SELECT role_id FROM user_roles WHERE user_id = $1;
            ",
            )
            .await?;

        let rows = tx.query(&stmt, &[&session.user.id]).await?;

        session.roles = rows.into_iter().map(|row| row.get(0)).collect();

        Ok(session)
    }
}

#[async_trait]
impl<Tls> UserDb for PostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    // TODO: clean up expired sessions?

    async fn logout(&self) -> Result<()> {
        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("DELETE FROM sessions WHERE id = $1;") // TODO: only invalidate session?
            .await?;

        conn.execute(&stmt, &[&self.session.id])
            .await
            .map_err(|_error| error::Error::LogoutFailed)?;
        Ok(())
    }

    async fn set_session_project(&self, project: ProjectId) -> Result<()> {
        // TODO: check permission

        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("UPDATE sessions SET project_id = $1 WHERE id = $2;")
            .await?;

        conn.execute(&stmt, &[&project, &self.session.id]).await?;

        Ok(())
    }

    async fn set_session_view(&self, view: STRectangle) -> Result<()> {
        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("UPDATE sessions SET view = $1 WHERE id = $2;")
            .await?;

        conn.execute(&stmt, &[&view, &self.session.id]).await?;

        Ok(())
    }

    async fn increment_quota_used(&self, user: &UserId, quota_used: u64) -> Result<()> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare(
                "
            UPDATE users SET 
                quota_available = quota_available - $1, 
                quota_used = quota_used + $1
            WHERE id = $2;",
            )
            .await?;

        conn.execute(&stmt, &[&(quota_used as i64), &user]).await?;

        Ok(())
    }

    async fn quota_used(&self) -> Result<u64> {
        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("SELECT quota_used FROM users WHERE id = $1;")
            .await?;

        let row = conn
            .query_one(&stmt, &[&self.session.user.id])
            .await
            .map_err(|_error| error::Error::InvalidSession)?;

        Ok(row.get::<usize, i64>(0) as u64)
    }

    async fn quota_used_by_user(&self, user: &UserId) -> Result<u64> {
        ensure!(
            self.session.user.id == *user || self.session.is_admin(),
            error::PermissionDenied
        );

        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("SELECT quota_used FROM users WHERE id = $1;")
            .await?;

        let row = conn
            .query_one(&stmt, &[&user])
            .await
            .map_err(|_error| error::Error::InvalidSession)?;

        Ok(row.get::<usize, i64>(0) as u64)
    }

    async fn quota_available(&self) -> Result<i64> {
        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("SELECT quota_available FROM users WHERE id = $1;")
            .await?;

        let row = conn
            .query_one(&stmt, &[&self.session.user.id])
            .await
            .map_err(|_error| error::Error::InvalidSession)?;

        Ok(row.get::<usize, i64>(0))
    }

    async fn quota_available_by_user(&self, user: &UserId) -> Result<i64> {
        ensure!(
            self.session.user.id == *user || self.session.is_admin(),
            error::PermissionDenied
        );

        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare("SELECT quota_available FROM users WHERE id = $1;")
            .await?;

        let row = conn
            .query_one(&stmt, &[&user])
            .await
            .map_err(|_error| error::Error::InvalidSession)?;

        Ok(row.get::<usize, i64>(0))
    }

    async fn update_quota_available_by_user(
        &self,
        user: &UserId,
        new_available_quota: i64,
    ) -> Result<()> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;
        let stmt = conn
            .prepare(
                "
            UPDATE users SET 
                quota_available = $1
            WHERE id = $2;",
            )
            .await?;

        conn.execute(&stmt, &[&(new_available_quota), &user])
            .await?;

        Ok(())
    }
}

#[async_trait]
impl<Tls> RoleDb for PostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    async fn add_role(&self, role_name: &str) -> Result<RoleId> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;

        let id = RoleId::new();

        let stmt = conn
            .prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")
            .await?;

        // TODO: map postgres error code to error::Error::RoleAlreadyExists

        conn.execute(&stmt, &[&id, &role_name]).await?;

        Ok(id)
    }

    async fn remove_role(&self, role_id: &RoleId) -> Result<()> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;

        let stmt = conn.prepare("DELETE FROM roles WHERE id = $1;").await?;

        let deleted = conn.execute(&stmt, &[&role_id]).await?;

        ensure!(deleted > 0, error::RoleDoesNotExist);

        Ok(())
    }

    async fn assign_role(&self, role_id: &RoleId, user_id: &UserId) -> Result<()> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")
            .await?;

        // TODO: map postgres error code to error::Error::RoleAlreadyAssigned, RoleDoesNotExist

        conn.execute(&stmt, &[&user_id, &role_id]).await?;

        Ok(())
    }

    async fn revoke_role(&self, role_id: &RoleId, user_id: &UserId) -> Result<()> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare("DELETE FROM user_roles WHERE user_id= $1 AND role_id = $2;")
            .await?;

        let deleted = conn.execute(&stmt, &[&user_id, &role_id]).await?;

        ensure!(deleted > 0, error::RoleNotAssigned);

        Ok(())
    }
}

1	use crate::contexts::SessionId;
2	use crate::error::Result;
3	use crate::pro::contexts::PostgresDb;
4	use crate::pro::permissions::{Role, RoleId};
5	use crate::pro::users::oidc::ExternalUserClaims;
6	use crate::pro::users::{
7	User, UserCredentials, UserDb, UserId, UserInfo, UserRegistration, UserSession,
8	};
9	use crate::projects::{ProjectId, STRectangle};
10	use crate::util::Identifier;
11	use crate::{error, pro::contexts::PostgresContext};
12	use async_trait::async_trait;
13
14	use bb8_postgres::{
15	tokio_postgres::tls::MakeTlsConnect, tokio_postgres::tls::TlsConnect, tokio_postgres::Socket,
16	};
17	use geoengine_datatypes::primitives::Duration;
18	use pwhash::bcrypt;
19	use snafu::ensure;
20	use uuid::Uuid;
21
22	use super::userdb::{RoleDb, UserAuth};
23
24	#[async_trait]
25	impl<Tls> UserAuth for PostgresContext<Tls>
26	where
27	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
28	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
29	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
30	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
31	{
32	// TODO: clean up expired sessions?
33
34	async fn register_user(&self, user: UserRegistration) -> Result<UserId> {	7✔
35	let mut conn = self.pool.get().await?;	7✔
36
37	let tx = conn.build_transaction().start().await?;	7✔
38
39	let user = User::from(user);	7✔
40
41	let stmt = tx	7✔
42	.prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")	7✔
43	.await?;	3✔
44	tx.execute(&stmt, &[&user.id, &user.email]).await?;	7✔
45
46	let stmt = tx	7✔
47	.prepare(	7✔
48	"INSERT INTO users (id, email, password_hash, real_name, quota_available, active) VALUES ($1, $2, $3, $4, $5, $6);",	7✔
49	)	7✔
50	.await?;	2✔
51
52	let quota_available =	7✔
53	crate::util::config::get_config_element::<crate::pro::util::config::User>()?	7✔
54	.default_available_quota;
55
56	tx.execute(	7✔
57	&stmt,	7✔
58	&[	7✔
59	&user.id,	7✔
60	&user.email,	7✔
61	&user.password_hash,	7✔
62	&user.real_name,	7✔
63	&quota_available,	7✔
64	&user.active,	7✔
65	],	7✔
66	)	7✔
67	.await?;	2✔
68
69	let stmt = tx	7✔
70	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	7✔
71	.await?;	2✔
72	tx.execute(&stmt, &[&user.id, &user.id]).await?;	7✔
73
74	let stmt = tx	7✔
75	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	7✔
76	.await?;	2✔
77	tx.execute(&stmt, &[&user.id, &Role::registered_user_role_id()])	7✔
78	.await?;	2✔
79
80	tx.commit().await?;	8✔
81
82	Ok(user.id)	7✔
83	}	14✔
84
85	async fn create_anonymous_session(&self) -> Result<UserSession> {	20✔
86	let mut conn = self.pool.get().await?;	20✔
87
88	let tx = conn.build_transaction().start().await?;	20✔
89
90	let user_id = UserId::new();	20✔
91
92	let stmt = tx	20✔
93	.prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")	20✔
94	.await?;	18✔
95	tx.execute(&stmt, &[&user_id, &format!("anonymous_user_{user_id}")])	20✔
96	.await?;	18✔
97
98	let quota_available =	20✔
99	crate::util::config::get_config_element::<crate::pro::util::config::User>()?	20✔
100	.default_available_quota;
101
102	let stmt = tx	20✔
103	.prepare("INSERT INTO users (id, quota_available, active) VALUES ($1, $2, TRUE);")	20✔
104	.await?;	18✔
105
106	tx.execute(&stmt, &[&user_id, &quota_available]).await?;	20✔
107
108	let stmt = tx	20✔
109	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	20✔
110	.await?;	16✔
111	tx.execute(&stmt, &[&user_id, &user_id]).await?;	20✔
112
113	let stmt = tx	20✔
114	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	20✔
115	.await?;	16✔
116	tx.execute(&stmt, &[&user_id, &Role::anonymous_role_id()])	20✔
117	.await?;	16✔
118
119	let session_id = SessionId::new();	20✔
120	let stmt = tx	20✔
121	.prepare(	20✔
122	"	20✔
123	INSERT INTO sessions (id, user_id, created, valid_until)	20✔
124	VALUES ($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3))	20✔
125	RETURNING created, valid_until;",	20✔
126	)	20✔
127	.await?;	16✔
128
129	// TODO: load from config
130	let session_duration = chrono::Duration::days(30);	20✔
131	let row = tx	20✔
132	.query_one(	20✔
133	&stmt,	20✔
134	&[	20✔
135	&session_id,	20✔
136	&user_id,	20✔
137	&(session_duration.num_seconds() as f64),	20✔
138	],	20✔
139	)	20✔
140	.await?;	17✔
141
142	tx.commit().await?;	20✔
143
144	Ok(UserSession {	20✔
145	id: session_id,	20✔
146	user: UserInfo {	20✔
147	id: user_id,	20✔
148	email: None,	20✔
149	real_name: None,	20✔
150	},	20✔
151	created: row.get(0),	20✔
152	valid_until: row.get(1),	20✔
153	project: None,	20✔
154	view: None,	20✔
155	roles: vec![user_id.into(), Role::anonymous_role_id()],	20✔
156	})	20✔
157	}	40✔
158
159	async fn login(&self, user_credentials: UserCredentials) -> Result<UserSession> {	22✔
160	let conn = self.pool.get().await?;	22✔
161	let stmt = conn	22✔
162	.prepare("SELECT id, password_hash, email, real_name FROM users WHERE email = $1;")	22✔
163	.await?;	11✔
164
165	let row = conn	22✔
166	.query_one(&stmt, &[&user_credentials.email])	22✔
167	.await	11✔
168	.map_err(\|_error\| error::Error::LoginFailed)?;	22✔
169
170	let user_id = UserId(row.get(0));	22✔
171	let password_hash = row.get(1);	22✔
172	let email = row.get(2);	22✔
173	let real_name = row.get(3);	22✔
174		22✔
175	if bcrypt::verify(user_credentials.password, password_hash) {	22✔
176	let session_id = SessionId::new();	22✔
177	let stmt = conn	22✔
178	.prepare(	22✔
179	"	22✔
180	INSERT INTO sessions (id, user_id, created, valid_until)	22✔
181	VALUES ($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3))	22✔
182	RETURNING created, valid_until;",	22✔
183	)	22✔
184	.await?;	10✔
185
186	// TODO: load from config
187	let session_duration = chrono::Duration::days(30);	22✔
188	let row = conn	22✔
189	.query_one(	22✔
190	&stmt,	22✔
191	&[	22✔
192	&session_id,	22✔
193	&user_id,	22✔
194	&(session_duration.num_seconds() as f64),	22✔
195	],	22✔
196	)	22✔
197	.await?;	22✔
198
199	let stmt = conn	22✔
200	.prepare("SELECT role_id FROM user_roles WHERE user_id = $1;")	22✔
201	.await?;	3✔
202
203	let rows = conn	22✔
204	.query(&stmt, &[&user_id])	22✔
205	.await	3✔
206	.map_err(\|_error\| error::Error::LoginFailed)?;	22✔
207
208	let roles = rows.into_iter().map(\|row\| row.get(0)).collect();	35✔
209		22✔
210	Ok(UserSession {	22✔
211	id: session_id,	22✔
212	user: UserInfo {	22✔
213	id: user_id,	22✔
214	email,	22✔
215	real_name,	22✔
216	},	22✔
217	created: row.get(0),	22✔
218	valid_until: row.get(1),	22✔
219	project: None,	22✔
220	view: None,	22✔
221	roles,	22✔
222	})	22✔
223	} else {
224	Err(error::Error::LoginFailed)	×
225	}
226	}	44✔
227
228	async fn login_external(	3✔
229	&self,	3✔
230	user: ExternalUserClaims,	3✔
231	duration: Duration,	3✔
232	) -> Result<UserSession> {	3✔
233	let mut conn = self.pool.get().await?;	3✔
234	let stmt = conn	3✔
235	.prepare("SELECT id, external_id, email, real_name FROM external_users WHERE external_id = $1;")	3✔
236	.await?;	3✔
237
238	let row = conn	3✔
239	.query_opt(&stmt, &[&user.external_id.to_string()])	3✔
240	.await	3✔
241	.map_err(\|_error\| error::Error::LoginFailed)?;	3✔
242
243	let user_id = match row {	3✔
244	Some(row) => UserId(row.get(0)),	2✔
245	None => {
246	let tx = conn.build_transaction().start().await?;	1✔
247
248	let user_id = UserId::new();	1✔
249
250	let stmt = tx	1✔
251	.prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")	1✔
252	.await?;	1✔
253	tx.execute(&stmt, &[&user_id, &user.email]).await?;	1✔
254
255	let quota_available =	1✔
256	crate::util::config::get_config_element::<crate::pro::util::config::User>()?	1✔
257	.default_available_quota;
258
259	//TODO: Inconsistent to hashmap implementation, where an external user is not part of the user database.
260	//TODO: A user might be able to login without external login using this (internal) id. Would be a problem with anonymous users as well.
261	let stmt = tx	1✔
262	.prepare(	1✔
263	"INSERT INTO users (id, quota_available, active) VALUES ($1, $2, TRUE);",	1✔
264	)	1✔
265	.await?;	1✔
266	tx.execute(&stmt, &[&user_id, &quota_available]).await?;	1✔
267
268	let stmt = tx	1✔
269	.prepare(	1✔
270	"INSERT INTO external_users (id, external_id, email, real_name, active) VALUES ($1, $2, $3, $4, $5);",	1✔
271	)	1✔
272	.await?;	1✔
273
274	tx.execute(	1✔
275	&stmt,	1✔
276	&[	1✔
277	&user_id,	1✔
278	&user.external_id.to_string(),	1✔
279	&user.email,	1✔
280	&user.real_name,	1✔
281	&true,	1✔
282	],	1✔
283	)	1✔
284	.await?;	1✔
285
286	let stmt = tx	1✔
287	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	1✔
288	.await?;	1✔
289	tx.execute(&stmt, &[&user_id, &user_id]).await?;	1✔
290
291	let stmt = tx	1✔
292	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	1✔
293	.await?;	1✔
294	tx.execute(&stmt, &[&user_id, &Role::registered_user_role_id()])	1✔
295	.await?;	1✔
296
297	tx.commit().await?;	1✔
298
299	user_id	1✔
300	}
301	};
302
303	let session_id = SessionId::new();	3✔
304	let stmt = conn	3✔
305	.prepare(	3✔
306	"	3✔
307	INSERT INTO sessions (id, user_id, created, valid_until)	3✔
308	VALUES ($1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + make_interval(secs:=$3))	3✔
309	RETURNING created, valid_until;",	3✔
310	)	3✔
311	.await?; //TODO: Check documentation if inconsistent to hashmap implementation - would happen if CURRENT_TIMESTAMP is called twice in postgres for a single query. Worked in tests.	3✔
312
313	let row = conn	3✔
314	.query_one(	3✔
315	&stmt,	3✔
316	&[&session_id, &user_id, &(duration.num_seconds() as f64)],	3✔
317	)	3✔
318	.await?;	3✔
319
320	let stmt = conn	3✔
321	.prepare("SELECT role_id FROM user_roles WHERE user_id = $1;")	3✔
322	.await?;	3✔
323
324	let rows = conn	3✔
325	.query(&stmt, &[&user_id])	3✔
326	.await	3✔
327	.map_err(\|_error\| error::Error::LoginFailed)?;	3✔
328
329	let roles = rows.into_iter().map(\|row\| row.get(0)).collect();	6✔
330		3✔
331	Ok(UserSession {	3✔
332	id: session_id,	3✔
333	user: UserInfo {	3✔
334	id: user_id,	3✔
335	email: Some(user.email.clone()),	3✔
336	real_name: Some(user.real_name.clone()),	3✔
337	},	3✔
338	created: row.get(0),	3✔
339	valid_until: row.get(1),	3✔
340	project: None,	3✔
341	view: None,	3✔
342	roles,	3✔
343	})	3✔
344	}	6✔
345
346	async fn user_session_by_id(&self, session: SessionId) -> Result<UserSession> {	13✔
347	let mut conn = self.pool.get().await?;	13✔
348
349	let tx = conn.build_transaction().start().await?;	13✔
350
351	let stmt = tx	13✔
352	.prepare(	13✔
353	"	13✔
354	SELECT	13✔
355	u.id,	13✔
356	u.email,	13✔
357	u.real_name,	13✔
358	s.created,	13✔
359	s.valid_until,	13✔
360	s.project_id,	13✔
361	s.view	13✔
362	FROM sessions s JOIN users u ON (s.user_id = u.id)	13✔
363	WHERE s.id = $1 AND CURRENT_TIMESTAMP < s.valid_until;",	13✔
364	)	13✔
365	.await?;	39✔
366
367	let row = tx	13✔
368	.query_one(&stmt, &[&session])	13✔
369	.await	9✔
370	.map_err(\|_error\| error::Error::InvalidSession)?;	13✔
371
372	let mut session = UserSession {	9✔
373	id: session,	9✔
374	user: UserInfo {	9✔
375	id: row.get(0),	9✔
376	email: row.get(1),	9✔
377	real_name: row.get(2),	9✔
378	},	9✔
379	created: row.get(3),	9✔
380	valid_until: row.get(4),	9✔
381	project: row.get::<usize, Option<Uuid>>(5).map(ProjectId),	9✔
382	view: row.get(6),	9✔
383	roles: vec![],	9✔
384	};	9✔
385
386	let stmt = tx	9✔
387	.prepare(	9✔
388	"	9✔
389	SELECT role_id FROM user_roles WHERE user_id = $1;	9✔
390	",	9✔
391	)	9✔
392	.await?;	6✔
393
394	let rows = tx.query(&stmt, &[&session.user.id]).await?;	9✔
395
396	session.roles = rows.into_iter().map(\|row\| row.get(0)).collect();	18✔
397		9✔
398	Ok(session)	9✔
399	}	26✔
400	}
401
402	#[async_trait]
403	impl<Tls> UserDb for PostgresDb<Tls>
404	where
405	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
406	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
407	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
408	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
409	{
410	// TODO: clean up expired sessions?
411
412	async fn logout(&self) -> Result<()> {	4✔
413	let conn = self.conn_pool.get().await?;	4✔
414	let stmt = conn	4✔
415	.prepare("DELETE FROM sessions WHERE id = $1;") // TODO: only invalidate session?	4✔
416	.await?;	3✔
417
418	conn.execute(&stmt, &[&self.session.id])	4✔
419	.await	4✔
420	.map_err(\|_error\| error::Error::LogoutFailed)?;	4✔
421	Ok(())	4✔
422	}	8✔
423
424	async fn set_session_project(&self, project: ProjectId) -> Result<()> {	2✔
425	// TODO: check permission
426
427	let conn = self.conn_pool.get().await?;	2✔
428	let stmt = conn	2✔
429	.prepare("UPDATE sessions SET project_id = $1 WHERE id = $2;")	2✔
430	.await?;	1✔
431
432	conn.execute(&stmt, &[&project, &self.session.id]).await?;	2✔
433
434	Ok(())	2✔
435	}	4✔
436
437	async fn set_session_view(&self, view: STRectangle) -> Result<()> {	2✔
438	let conn = self.conn_pool.get().await?;	2✔
439	let stmt = conn	2✔
440	.prepare("UPDATE sessions SET view = $1 WHERE id = $2;")	2✔
441	.await?;	1✔
442
443	conn.execute(&stmt, &[&view, &self.session.id]).await?;	2✔
444
445	Ok(())	2✔
446	}	4✔
447
448	async fn increment_quota_used(&self, user: &UserId, quota_used: u64) -> Result<()> {	4✔
449	ensure!(self.session.is_admin(), error::PermissionDenied);	4✔
450
451	let conn = self.conn_pool.get().await?;	4✔
452	let stmt = conn	4✔
453	.prepare(	4✔
454	"	4✔
455	UPDATE users SET	4✔
456	quota_available = quota_available - $1,	4✔
457	quota_used = quota_used + $1	4✔
458	WHERE id = $2;",	4✔
459	)	4✔
460	.await?;	4✔
461
462	conn.execute(&stmt, &[&(quota_used as i64), &user]).await?;	4✔
463
464	Ok(())	4✔
465	}	8✔
466
467	async fn quota_used(&self) -> Result<u64> {	2✔
468	let conn = self.conn_pool.get().await?;	2✔
469	let stmt = conn	2✔
470	.prepare("SELECT quota_used FROM users WHERE id = $1;")	2✔
471	.await?;	2✔
472
473	let row = conn	2✔
474	.query_one(&stmt, &[&self.session.user.id])	2✔
475	.await	2✔
476	.map_err(\|_error\| error::Error::InvalidSession)?;	2✔
477
478	Ok(row.get::<usize, i64>(0) as u64)	2✔
479	}	4✔
480
481	async fn quota_used_by_user(&self, user: &UserId) -> Result<u64> {	×
482	ensure!(	×
483	self.session.user.id == *user \|\| self.session.is_admin(),	×
484	error::PermissionDenied	×
485	);
486
487	let conn = self.conn_pool.get().await?;	×
488	let stmt = conn	×
489	.prepare("SELECT quota_used FROM users WHERE id = $1;")	×
490	.await?;	×
491
492	let row = conn	×
493	.query_one(&stmt, &[&user])	×
494	.await	×
495	.map_err(\|_error\| error::Error::InvalidSession)?;	×
496
497	Ok(row.get::<usize, i64>(0) as u64)	×
498	}	×
499
500	async fn quota_available(&self) -> Result<i64> {	4✔
501	let conn = self.conn_pool.get().await?;	4✔
502	let stmt = conn	4✔
503	.prepare("SELECT quota_available FROM users WHERE id = $1;")	4✔
504	.await?;	2✔
505
506	let row = conn	4✔
507	.query_one(&stmt, &[&self.session.user.id])	4✔
508	.await	2✔
509	.map_err(\|_error\| error::Error::InvalidSession)?;	4✔
510
511	Ok(row.get::<usize, i64>(0))	4✔
512	}	8✔
513
514	async fn quota_available_by_user(&self, user: &UserId) -> Result<i64> {	2✔
515	ensure!(	2✔
516	self.session.user.id == *user \|\| self.session.is_admin(),	2✔
517	error::PermissionDenied	×
518	);
519
520	let conn = self.conn_pool.get().await?;	2✔
521	let stmt = conn	2✔
522	.prepare("SELECT quota_available FROM users WHERE id = $1;")	2✔
523	.await?;	×
524
525	let row = conn	2✔
526	.query_one(&stmt, &[&user])	2✔
527	.await	×
528	.map_err(\|_error\| error::Error::InvalidSession)?;	2✔
529
530	Ok(row.get::<usize, i64>(0))	2✔
531	}	4✔
532
533	async fn update_quota_available_by_user(	2✔
534	&self,	2✔
535	user: &UserId,	2✔
536	new_available_quota: i64,	2✔
537	) -> Result<()> {	2✔
538	ensure!(self.session.is_admin(), error::PermissionDenied);	2✔
539
540	let conn = self.conn_pool.get().await?;	2✔
541	let stmt = conn	2✔
542	.prepare(	2✔
543	"	2✔
544	UPDATE users SET	2✔
545	quota_available = $1	2✔
546	WHERE id = $2;",	2✔
547	)	2✔
548	.await?;	×
549
550	conn.execute(&stmt, &[&(new_available_quota), &user])	2✔
551	.await?;	2✔
552
553	Ok(())	2✔
554	}	4✔
555	}
556
557	#[async_trait]
558	impl<Tls> RoleDb for PostgresDb<Tls>
559	where
560	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
561	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
562	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
563	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
564	{
565	async fn add_role(&self, role_name: &str) -> Result<RoleId> {	1✔
566	ensure!(self.session.is_admin(), error::PermissionDenied);	1✔
567
568	let conn = self.conn_pool.get().await?;	1✔
569
570	let id = RoleId::new();	1✔
571
572	let stmt = conn	1✔
573	.prepare("INSERT INTO roles (id, name) VALUES ($1, $2);")	1✔
574	.await?;	×
575
576	// TODO: map postgres error code to error::Error::RoleAlreadyExists
577
578	conn.execute(&stmt, &[&id, &role_name]).await?;	1✔
579
580	Ok(id)	1✔
581	}	2✔
582
583	async fn remove_role(&self, role_id: &RoleId) -> Result<()> {	1✔
584	ensure!(self.session.is_admin(), error::PermissionDenied);	1✔
585
586	let conn = self.conn_pool.get().await?;	1✔
587
588	let stmt = conn.prepare("DELETE FROM roles WHERE id = $1;").await?;	1✔
589
590	let deleted = conn.execute(&stmt, &[&role_id]).await?;	1✔
591
592	ensure!(deleted > 0, error::RoleDoesNotExist);	1✔
593
594	Ok(())	1✔
595	}	2✔
596
597	async fn assign_role(&self, role_id: &RoleId, user_id: &UserId) -> Result<()> {	2✔
598	ensure!(self.session.is_admin(), error::PermissionDenied);	2✔
599
600	let conn = self.conn_pool.get().await?;	2✔
601
602	let stmt = conn	2✔
603	.prepare("INSERT INTO user_roles (user_id, role_id) VALUES ($1, $2);")	2✔
604	.await?;	×
605
606	// TODO: map postgres error code to error::Error::RoleAlreadyAssigned, RoleDoesNotExist
607
608	conn.execute(&stmt, &[&user_id, &role_id]).await?;	2✔
609
610	Ok(())	2✔
611	}	4✔
612
613	async fn revoke_role(&self, role_id: &RoleId, user_id: &UserId) -> Result<()> {	1✔
614	ensure!(self.session.is_admin(), error::PermissionDenied);	1✔
615
616	let conn = self.conn_pool.get().await?;	1✔
617
618	let stmt = conn	1✔
619	.prepare("DELETE FROM user_roles WHERE user_id= $1 AND role_id = $2;")	1✔
620	.await?;	×
621
622	let deleted = conn.execute(&stmt, &[&user_id, &role_id]).await?;	1✔
623
624	ensure!(deleted > 0, error::RoleNotAssigned);	1✔
625
626	Ok(())	1✔
627	}	2✔
628	}

geo-engine / geoengine / 4594239788

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous