# User Permission Agent
This report documents the implementation of a full **Role-Based Access Control (RBAC)** system for a LINE chatbot application using **Semantic Kernel Agents**. The system implements two-factor authentication via email verification and controls access to functions based on user role levels.
This technical report focuses on two main areas: **authentication** and **authorization**. Each section provides detailed implementation steps.
---
## System Architecture Overview
The authorization system consists of two main components:
1. **Authentication Layer**: Email-based user verification with time-limited verification codes.
2. **Authorization Layer**: Restrict Kernel Function usage based on user roles.
The system integrates with the LINE messaging platform and uses MongoDB for persistent storage of user data and verification records.
---
## Authentication
### Authentication Flow Diagram
### Core Authentication Components
#### 1. Database Collections
* **`employee`**: Stores employee records and LINE ID bindings
```javascript
{
"_id" : ObjectId("686b6ecca9eca0c6d14c99da"),
"employee_id" : "A15",
"name" : "Woody",
"email" : "woodychang891121@gmail.com",
"store_id" : "S003",
"status" : "active",
"hire_date" : ISODate("2025-07-07T00:00:00.000+0000"),
"shift_type" : "Part-time",
"role_level" : "Owner",
"line_id" : "U92fabcda9d16bbbfdeae89ba26fcbe36"
}
```
* **`email_verification`**: Manages verification codes and expiration times
```javascript
{
"_id" : ObjectId("686b8405a47f80045033fd2b"),
"email" : "woodychang891121@gmail.com",
"code" : "244251",
"expires_at" : ISODate("2025-07-08T06:12:09.221+0000"),
"verified" : true,
"line_id" : "U92fabcda9d16bbbfdeae89ba26fcbe36"
}
```
#### 2. Authentication Agent (`auth_agent.py`)
The authentication agent is the entry point for user verification, built using Semantic Kernel’s `ChatCompletionAgent`.
```python
auth_agent = ChatCompletionAgent(
service=service,
name="AuthAgent",
instructions=AUTH_PROMPT,
plugins=[auth_plugin]
)
```
#### 3. Authorization Plugin (`auth_plugin` class)
The plugin implements core authentication and authorization logic through kernel functions.
It provides all necessary functions to the authentication bot, ==**allowing the AI agent to automatically determine what information the user must provide, and identify which step of the authentication process the user is in**==.
---
### Authentication Implementation Details
The following tools belong to the authentication AI agent. The agent automatically selects the appropriate tool depending on the user’s response.
#### Step 1: Check if email exists
```python
@kernel_function(name="find_user_by_email", description="Check if the email exists in the employee database")
def find_user_by_email(self, email: Annotated[str, "user email, it should look something like example@domain.com.tw or example@domain.com"]) -> str:
employee = employee_collection.find_one({"email": email})
if employee is not None:
return "User email exists in the database"
else:
return "User email is not registered in the database"
```
#### Step 2: Send verification email
```python
@kernel_function(name="send_verification_email", description="Send a verification code to user email")
def send_verification_email(self, email: str) -> str:
code = str(random.randint(100000, 999999))
expires_at = datetime.now() + timedelta(minutes=5)
verification_collection.update_one(
{"email": email},
{"$set": {
"code": code,
"expires_at": expires_at,
"verified": False
}},
upsert=True
)
return AuthPlugin.send_email_verification_code(email, code)
```
```python
def send_email_verification_code(to_email: str, code: str) -> str:
subject = "Your Verification Code"
body = f"Hello,\n\nYour verification code is: {code}\nThis code will expire in 5 minutes.\n\nThanks."
# MIME setup
msg = MIMEMultipart()
msg["From"] = AuthPlugin.SENDER_EMAIL
msg["To"] = to_email
msg["Subject"] = subject
msg.attach(MIMEText(body, "plain"))
try:
server = smtplib.SMTP(AuthPlugin.SMTP_SERVER, AuthPlugin.SMTP_PORT)
server.starttls()
server.login(AuthPlugin.SENDER_EMAIL, AuthPlugin.SENDER_PASSWORD)
server.sendmail(AuthPlugin.SENDER_EMAIL, to_email, msg.as_string())
server.quit()
return "Verification email sent"
except Exception as e:
print("Failed to send email:", e)
return "System failed to send email"
def get_user_info_by_line_id(user_id: str):
employee = employee_collection.find_one({"line_id": user_id})
if employee:
return {
"user_name": employee.get("name"),
"role": employee.get("role_level")
}
else:
return None
```
**Verification Code Features:**
* 6-digit random number
* 5-minute expiration
* Sent via Gmail SMTP service
#### Step 3: Code validation
```python
@kernel_function(name="validate_code", description="Check if the verification code is correct")
def validate_code(self, line_id: str, code: str) -> str:
record = verification_collection.find_one({"code": code})
if not record:
return "User entered an invalid code. Ask user to confirm."
if datetime.now() > record["expires_at"]:
return "Verification code expired. Ask user if they need a new one."
verification_collection.update_one({"code": code}, {"$set": {"verified": True, "line_id": line_id}})
return "User verification code validated"
```
#### Step 4: LINE ID binding
```python
@kernel_function(name="bind_line_id", description="Bind LINE ID to the employee record by email")
def bind_line_id(self, email: str, line_id: str) -> str:
result = employee_collection.update_one(
{"email": email},
{"$set": {"line_id": line_id}}
)
if result.modified_count == 1:
return "User LINE ID successfully registered"
else:
return "User LINE ID could not be registered. Check authentication steps."
```
---
### Main Flow Control
```python
user_id = event.source.user_id
user_message = event.message.text
thread = thread_manager.get_thread(user_id)
# Check if user ID is already in database and bound to email
if not AuthPlugin.line_id_in_db(user_id):
# Start authentication via AuthAgent
verification_response = await auth_agent.get_response(
messages=user_message + f"LINE ID: {user_id}",
thread=thread
)
verification_response = verification_response.items[0].text
line_text_response(event, verification_response)
return
# Main flow: route user query to appropriate agent
```
### Auxiliary Functions
#### LINE ID Database Check
```python
# Check if user’s LINE ID exists in database
def line_id_in_db(line_id: str) -> bool:
employee = employee_collection.find_one({"line_id": line_id})
return employee is not None
```
---
## Security Considerations
### 1. Verification Code Security
* **Time-limited**: 5-minute expiration
* **Randomness**: 6-digit random number
* **One-time use**: Marked as verified after use
### 2. Email Security
* **Gmail SMTP**: Use app passwords instead of standard login
* **TLS encryption**: STARTTLS for secure email transmission
* **Environment variables**: Store sensitive info in environment variables
---
## Authorization Control
### Kernel Function (Tool) Level Authorization
```python
role_level = ['Operations', 'Marketing', 'Procurement', 'HR', 'Owner']
# Tool access control map
function_role_access_map = {
'search_related_product_info': {
"role": ['Operations','Marketing', 'Procurement'],
"warning": "❌ You do not have permission to use this function. Only Operations, Marketing, and Procurement can query product information."
},
'get_ranking': {
"role": ['Operations','Marketing', 'Procurement'],
"warning": "❌ You do not have permission to use this function. Only Operations, Marketing, and Procurement can query sales ranking."
},
'get_sales_data': {
'role': ['Operations','Marketing'],
'warning': "❌ You do not have permission to use this function. Only Operations and Marketing can query sales data."
}
```
Auxiliary function for access control:
```python
def check_user_role_access(function_name: str, user_role: str):
print(user_role)
if user_role not in function_role_access_map[function_name]['role']:
return function_role_access_map[function_name]['warning']
return None
```
```python
def get_product_information(
self,
user_input:Annotated[str, "Full product info query derived from context"],
user_role: Annotated[str, f"User role from system. Roles include {role_level}"]
)-> PluginResponse:
# Check if user has permission
if check_user_role_access('search_related_product_info', user_role):
return check_user_role_access('search_related_product_info', user_role)
# Additional kernel function logic
...
```
---
## Scalability Considerations
This user permission system can serve as a standalone extensible module. Designed as a reusable plugin, it can be easily integrated into other projects. Example structure:
```
project_root/
├── .history
├── app
├── auth/ # Authentication module
│ ├── __pycache__
│ ├── prompts/
│ │ └── auth.txt
│ ├── __init__.py
│ ├── auth_agents.py
│ └── auth_plugin.py
├── market_utils # Marketing Agent module
└── manager_utils # Management Agent module
```
### Adding a New Role
1. Update `role_level` list
2. Add access rules in `function_role_access_map`
3. Update user role info in the database
### Adding a New Function
1. Define access rules in `function_role_access_map`
2. Implement authorization check in the corresponding kernel function
---
## Conclusion
This system implements full RBAC with **email verification** and **function-level access control**. The design is secure and extensible, allowing easy addition of new roles and features.
Through Semantic Kernel’s architecture, the system integrates seamlessly with chatbot workflows, providing smooth user experiences. It is recommended to perform thorough testing before production deployment, especially for email sending and database connectivity.
Sign in with Wallet
Connect another wallet