4462097035

Build # 4462097035

Build Type

push

github

Committed by GitHub

Commit Message

Merge #756

Run Details

4251 of 4251 new or added lines in 67 files covered. (100.0%)

92094 of 105555 relevant lines covered (87.25%)

75190.51 hits per line

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

95.28

/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::user_input::Validated;
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: Validated<UserRegistration>) -> Result<UserId> {
        let mut conn = self.pool.get().await?;

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

        let user = User::from(user.user_input);

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

geo-engine / geoengine / 4462097035

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