给IT tools加Oauth的代码示例

IT tools是Vue的纯前端项目。其实可以接入SSO。


https://appstore.lazycat.cloud/#/shop/detail/iamxiaoe.lzcapp.ittools

修改路由器守卫：涵盖了动态路由生成、组件懒加载、基于 token 的登录鉴权和未登录用户的自动重定向。所有受保护的页面都会在跳转前通过 API 校验 token 的有效性，确保只有合法用户才能访问。同时，我们为工具类页面统一指定了专属布局
```jsx
import { createRouter, createWebHistory } from 'vue-router';
import { layouts } from './layouts/index';
import HomePage from './pages/Home.page.vue';
import NotFound from './pages/404.page.vue';
import { tools } from './tools';
import { config } from './config';
import { routes as demoRoutes } from './ui/demo/demo.routes';
import AuthCallback from './pages/AuthCallback.page.vue';
import Login from './pages/Login.page.vue';
import axios, { AxiosResponse } from 'axios';

const toolsRoutes = tools.map(({ path, name, component, ...config }) => ({
  path,
  name,
  component,
  meta: { requiresAuth: true, isTool: true, layout: layouts.toolLayout, name, ...config },
}));
const toolsRedirectRoutes = tools
  .filter(({ redirectFrom }) => redirectFrom && redirectFrom.length > 0)
  .flatMap(
    ({ path, redirectFrom }) => redirectFrom?.map(redirectSource => ({ path: redirectSource, redirect: path })) ?? [],
  );

const router = createRouter({
  history: createWebHistory(config.app.baseUrl),
  routes: [
    {
      path: '/',
      name: 'home',
      component: HomePage,
      meta: { requiresAuth: true }
    },
    {
      path: '/about',
      name: 'about',
      component: () => import('./pages/About.vue'),
      meta: { requiresAuth: true }
    },
    {
      path: '/login',
      name: 'Login',
      component: Login
    },
    {
      path: '/auth/callback',
      name: 'AuthCallback',
      component: AuthCallback,
    },
    ...toolsRoutes,
    ...toolsRedirectRoutes,
    ...(config.app.env === 'development' ? demoRoutes : []),
    { path: '/:pathMatch(.*)*', name: 'NotFound', component: NotFound },
  ],
});

router.beforeEach((to, from, next) => {
  const token = localStorage.getItem('access_token');

  if (to.matched.some(record => record.meta.requiresAuth)) {
    if (!token) {
      // 如果没有token，重定向到登录页面
      next({
        path: '/login',
        query: { redirect: to.fullPath }
      });
    } else {
      // 如果有token，检查token的有效性
      axios.post(`${import.meta.env.VITE_APP_API_URL}:8000/session`,  {},{
        headers: {
          'Authorization': `Bearer ${token}`
        }
      })
      .then(response => {
        // 根据响应判断token是否有效
        if (response.data.error_code === 200 && response.data.message === 'Valid session') {
          // Token有效，放行
          next();
        } else {
          // Token无效，清除localStorage中的token并重定向到登录页面
          localStorage.removeItem('access_token');
          next({
            path: '/login',
            query: { redirect: to.fullPath }
          });
        }
      })
      .catch(error => {
        // 请求出错，清除localStorage中的token并重定向到登录页面
        localStorage.removeItem('access_token');
        next({
          path: '/login',
          query: { redirect: to.fullPath }
        });
      });
    }
  } else {
    // 如果路由不需要认证，直接放行
    next();
  }
});

export default router;
```

在实现单点登录（SSO）时，我们设计了一个专门的跳转页组件。当用户访问该页面时，前端会自动拼接好后端的登录地址（带上回调参数 redirect_uri），并立即将浏览器重定向到该地址，发起 SSO 登录流程。这个组件无需展示 UI，只负责发起跳转，逻辑简单但至关重要，是整个认证链路中的入口节点。



```jsx

  
    SSO...
   -->



import axios from 'axios';

export default {
  mounted() {
    this.redirectToSSO();
  },
  methods: {
    redirectToSSO() {
      console.log(`API: ${import.meta.env.VITE_APP_API_URL}`)
      const ssoLoginUrl = `${import.meta.env.VITE_APP_API_URL||"na"}:8000/login?redirect_uri=${import.meta.env.VITE_APP_API_URL||"na"}/auth/callback`;
      window.location.href = ssoLoginUrl;
    }
  }
}


```

前端回调

当用户完成 SSO 授权并跳转回前端时，我们通过一个专门的回调页面组件捕获 URL 中的授权码 code。组件加载时自动执行鉴权逻辑：

从 code 中构造后端请求，调用 /callback 获取 access_token

成功后将 token 写入 localStorage，并跳转回首页

若失败或缺少授权码，则重定向回登录页重新发起认证

这一过程实现了 OAuth2 标准中的「授权码交换令牌」流程，是前后端 SSO 登录闭环的关键一环。

```jsx

  
    SSO...
  



import axios from 'axios'; 

export default {
  created() {
    this.handleAuthentication();
  },
  methods: {
    handleAuthentication() {
      console.log("handleAuthentication hook")
      const code = new URLSearchParams(this.$route.query).get('code');
      if (code) {
        // console.log(1, code)
        axios.post(`${import.meta.env.VITE_APP_API_URL||"na"}:8000/callback?code=${code}`,)
          .then(response => {
            // console.log("response",response)
            localStorage.setItem('access_token', response.data.access_token);
            this.$router.push('/');
          })
          .catch(() => {
            this.$router.push('/login');
          });
      } else {
        this.$router.push('/login');
      }
    }
  }
}

```

后端跳转


为实现前后端分离的单点登录（SSO）功能，我们基于 FastAPI 构建了轻量级认证服务。后端会根据环境自动从 AWS Secrets Manager 加载 SSO 配置信息，避免敏感信息硬编码。整个登录流程遵循 OAuth2 协议，包括：

登录接口 /login 会构造并重定向用户到授权服务器

用户授权后由 /callback 接收授权码并换取访问令牌

前端持有的 token 可通过 /session 接口进行二次校验，确保权限合法

整套认证链路简洁、模块化，适用于支持 GitHub、Auth0 等主流 SSO 平台的企业级项目。


```python
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.responses import RedirectResponse, JSONResponse
import requests
import os
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv
import boto3
from botocore.exceptions import ClientError
import json
from fastapi import FastAPI, Request, Header

app = FastAPI()
load_dotenv()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  
    allow_credentials=True,
    allow_methods=["*"],  
    allow_headers=["*"], 
)

secrets_client = boto3.client('secretsmanager')

def get_secret(environment):

    secret_name = f"{environment}/xxxx"
    region_name = "cn-northwest-1"

    # Create a Secrets Manager client
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name
    )

    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=secret_name
        )
    except ClientError as e:
        raise e

    secret = get_secret_value_response['SecretString']
    secret = json.loads(secret)

    return secret

environment = os.getenv('ENV', 'dev')
secrets = get_secret(environment)

CLIENT_ID = secrets.get('CLIENT_ID', 'na')
CLIENT_SECRET = secrets.get('CLIENT_SECRET', 'na')
AUTHORIZATION_BASE_URL = secrets.get('AUTHORIZATION_BASE_URL', 'na')
TOKEN_URL = secrets.get('TOKEN_URL', 'na')
REDIRECT_URI = secrets.get('REDIRECT_URI', 'na')
SESSION_URL = secrets.get('SESSION_URL', 'na')

@app.get("/")
def read_root():
    return {
        "ENVIRONMENT": environment,
        "AUTHORIZATION_BASE_URL": AUTHORIZATION_BASE_URL
        }

@app.get("/login")
def login():
    authorization_url = (
        f"{AUTHORIZATION_BASE_URL}?response_type=code&client_id={CLIENT_ID}&redirect_uri={REDIRECT_URI}"
    )
    return RedirectResponse(authorization_url)

@app.post("/callback")
def callback(request: Request):
    code = request.query_params.get('code')
    print("code",code)
    if not code:
        raise HTTPException(status_code=400, detail="Missing authorization code")

    token_data = {
        'grant_type': 'authorization_code',
        'code': code,
        'redirect_uri': REDIRECT_URI,
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
    }

    try:
        token_response = requests.post(TOKEN_URL, data=token_data)
        token_response.raise_for_status()
        token_json = token_response.json()
        access_token = token_json.get('access_token')
    except requests.RequestException as e:
        raise HTTPException(status_code=400, detail=str(e))

    return JSONResponse(content={"access_token": access_token})

@app.post("/session")
def auth_session(Authorization: str = Header(default=None)):
    headers = {'Authorization': f'{Authorization}'}

    try:
        token_response = requests.get(SESSION_URL, headers=headers)
        token_response.raise_for_status()
        token_json = token_response.json()
        msg = token_json.get("message")
        if msg == "Valid session":
            return token_json
    except requests.RequestException as e:
        raise HTTPException(status_code=400, detail=str(e))

if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host='0.0.0.0', port=8000)
```